{ "cells": [ { "cell_type": "markdown", "id": "c4419168-c0e6-4a65-b56e-8454c42060ac", "metadata": { "tags": [] }, "source": [ "### 0. Spark Setup" ] }, { "cell_type": "code", "execution_count": 1, "id": "32bd7cdd-8504-4a54-a461-244bf7878d2a", "metadata": {}, "outputs": [], "source": [ "import os\n", "os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.1,org.apache.spark:spark-avro_2.12:3.3.1 pyspark-shell'" ] }, { "cell_type": "code", "execution_count": 2, "id": "3aab2a7e-a685-4925-9c9a-b5adf201af77", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "your 131072x1 screen size is bogus. expect trouble\n", "24/03/11 00:28:48 WARN Utils: Your hostname, Cinders resolves to a loopback address: 127.0.1.1; using 172.17.156.62 instead (on interface eth0)\n", "24/03/11 00:28:48 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ ":: loading settings :: url = jar:file:/home/ellabelle/spark/spark-3.5.1-bin-hadoop3/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "Ivy Default Cache set to: /home/ellabelle/.ivy2/cache\n", "The jars for the packages stored in: /home/ellabelle/.ivy2/jars\n", "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n", "org.apache.spark#spark-avro_2.12 added as a dependency\n", ":: resolving dependencies :: org.apache.spark#spark-submit-parent-0c8615d6-fa19-46ec-942b-46e9fe0012aa;1.0\n", "\tconfs: [default]\n", "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.3.1 in central\n", "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.1 in central\n", "\tfound org.apache.kafka#kafka-clients;2.8.1 in central\n", "\tfound org.lz4#lz4-java;1.8.0 in central\n", "\tfound org.xerial.snappy#snappy-java;1.1.8.4 in central\n", "\tfound org.slf4j#slf4j-api;1.7.32 in central\n", "\tfound org.apache.hadoop#hadoop-client-runtime;3.3.2 in central\n", "\tfound org.spark-project.spark#unused;1.0.0 in central\n", "\tfound org.apache.hadoop#hadoop-client-api;3.3.2 in central\n", "\tfound commons-logging#commons-logging;1.1.3 in central\n", "\tfound com.google.code.findbugs#jsr305;3.0.0 in central\n", "\tfound org.apache.commons#commons-pool2;2.11.1 in central\n", "\tfound org.apache.spark#spark-avro_2.12;3.3.1 in central\n", "\tfound org.tukaani#xz;1.8 in central\n", ":: resolution report :: resolve 328ms :: artifacts dl 13ms\n", "\t:: modules in use:\n", "\tcom.google.code.findbugs#jsr305;3.0.0 from central in [default]\n", "\tcommons-logging#commons-logging;1.1.3 from central in [default]\n", "\torg.apache.commons#commons-pool2;2.11.1 from central in [default]\n", "\torg.apache.hadoop#hadoop-client-api;3.3.2 from central in [default]\n", "\torg.apache.hadoop#hadoop-client-runtime;3.3.2 from central in [default]\n", "\torg.apache.kafka#kafka-clients;2.8.1 from central in [default]\n", "\torg.apache.spark#spark-avro_2.12;3.3.1 from central in [default]\n", "\torg.apache.spark#spark-sql-kafka-0-10_2.12;3.3.1 from central in [default]\n", "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.1 from central in [default]\n", "\torg.lz4#lz4-java;1.8.0 from central in [default]\n", "\torg.slf4j#slf4j-api;1.7.32 from central in [default]\n", "\torg.spark-project.spark#unused;1.0.0 from central in [default]\n", "\torg.tukaani#xz;1.8 from central in [default]\n", "\torg.xerial.snappy#snappy-java;1.1.8.4 from central in [default]\n", "\t---------------------------------------------------------------------\n", "\t| | modules || artifacts |\n", "\t| conf | number| search|dwnlded|evicted|| number|dwnlded|\n", "\t---------------------------------------------------------------------\n", "\t| default | 14 | 0 | 0 | 0 || 14 | 0 |\n", "\t---------------------------------------------------------------------\n", ":: retrieving :: org.apache.spark#spark-submit-parent-0c8615d6-fa19-46ec-942b-46e9fe0012aa\n", "\tconfs: [default]\n", "\t0 artifacts copied, 14 already retrieved (0kB/8ms)\n", "24/03/11 00:28:50 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n", "Setting default log level to \"WARN\".\n", "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n", "24/03/11 00:28:51 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.\n" ] } ], "source": [ "from pyspark.sql import SparkSession\n", "import pyspark.sql.types as T\n", "import pyspark.sql.functions as F\n", "\n", "spark = SparkSession \\\n", " .builder \\\n", " .appName(\"Spark-Notebook\") \\\n", " .getOrCreate()" ] }, { "cell_type": "markdown", "id": "6f4b62fa-b3ce-4a1b-a1f4-2ed332a0d55a", "metadata": { "tags": [] }, "source": [ "### 1. Reading from Kafka Stream\n", "\n", "through `readStream`" ] }, { "cell_type": "markdown", "id": "f491fa45-4471-4bc5-92f7-48081f687140", "metadata": {}, "source": [ "#### 1.1 Raw Kafka Stream" ] }, { "cell_type": "code", "execution_count": 3, "id": "82c25cb2-2599-4f9b-8849-967fbb604a44", "metadata": { "tags": [] }, "outputs": [], "source": [ "# default for startingOffsets is \"latest\"\n", "df_kafka_raw = spark \\\n", " .readStream \\\n", " .format(\"kafka\") \\\n", " .option(\"kafka.bootstrap.servers\", \"localhost:9092,broker:29092\") \\\n", " .option(\"subscribe\", \"rides_csv\") \\\n", " .option(\"startingOffsets\", \"earliest\") \\\n", " .option(\"checkpointLocation\", \"checkpoint\") \\\n", " .load()" ] }, { "cell_type": "code", "execution_count": 4, "id": "d9149ccd-69b2-4f5b-afc0-43567673c634", "metadata": { "tags": [] }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- key: binary (nullable = true)\n", " |-- value: binary (nullable = true)\n", " |-- topic: string (nullable = true)\n", " |-- partition: integer (nullable = true)\n", " |-- offset: long (nullable = true)\n", " |-- timestamp: timestamp (nullable = true)\n", " |-- timestampType: integer (nullable = true)\n", "\n" ] } ], "source": [ "df_kafka_raw.printSchema()" ] }, { "cell_type": "markdown", "id": "62e5e753-89c7-460f-a8be-16868ce5c680", "metadata": { "tags": [] }, "source": [ "#### 1.2 Encoded Kafka Stream" ] }, { "cell_type": "code", "execution_count": 5, "id": "0b745eed-7d74-421e-8e4b-c8343fda4de3", "metadata": { "tags": [] }, "outputs": [], "source": [ "df_kafka_encoded = df_kafka_raw.selectExpr(\"CAST(key AS STRING)\",\"CAST(value AS STRING)\")" ] }, { "cell_type": "code", "execution_count": 6, "id": "6839addc-c7c0-4117-8c9c-d2cd59cbf136", "metadata": { "tags": [] }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- key: string (nullable = true)\n", " |-- value: string (nullable = true)\n", "\n" ] } ], "source": [ "df_kafka_encoded.printSchema()" ] }, { "cell_type": "markdown", "id": "6749c4de-6f80-4b91-b2b8-b2968c761d75", "metadata": {}, "source": [ "#### 1.3 Structure Streaming DataFrame" ] }, { "cell_type": "code", "execution_count": 7, "id": "ca20ae37-49f0-421f-9859-73fac8d4ca45", "metadata": { "tags": [] }, "outputs": [], "source": [ "def parse_ride_from_kafka_message(df_raw, schema):\n", " \"\"\" take a Spark Streaming df and parse value col based on , return streaming df cols in schema \"\"\"\n", " assert df_raw.isStreaming is True, \"DataFrame doesn't receive streaming data\"\n", "\n", " df = df_raw.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\")\n", "\n", " # split attributes to nested array in one Column\n", " col = F.split(df['value'], ', ')\n", "\n", " # expand col to multiple top-level columns\n", " for idx, field in enumerate(schema):\n", " df = df.withColumn(field.name, col.getItem(idx).cast(field.dataType))\n", " return df.select([field.name for field in schema])" ] }, { "cell_type": "code", "execution_count": 8, "id": "e1737bd0-146f-4ee2-a70f-a4657af5bbc6", "metadata": { "tags": [] }, "outputs": [], "source": [ "ride_schema = T.StructType(\n", " [T.StructField(\"vendor_id\", T.IntegerType()),\n", " T.StructField('tpep_pickup_datetime', T.TimestampType()),\n", " T.StructField('tpep_dropoff_datetime', T.TimestampType()),\n", " T.StructField(\"passenger_count\", T.IntegerType()),\n", " T.StructField(\"trip_distance\", T.FloatType()),\n", " T.StructField(\"payment_type\", T.IntegerType()),\n", " T.StructField(\"total_amount\", T.FloatType()),\n", " ])" ] }, { "cell_type": "code", "execution_count": 9, "id": "ae2ce896-f54b-4166-b01f-b5532ab292fe", "metadata": { "tags": [] }, "outputs": [], "source": [ "df_rides = parse_ride_from_kafka_message(df_raw=df_kafka_raw, schema=ride_schema)" ] }, { "cell_type": "code", "execution_count": 10, "id": "cd848228-97c5-4325-8457-97f35e533cd8", "metadata": { "tags": [] }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "root\n", " |-- vendor_id: integer (nullable = true)\n", " |-- tpep_pickup_datetime: timestamp (nullable = true)\n", " |-- tpep_dropoff_datetime: timestamp (nullable = true)\n", " |-- passenger_count: integer (nullable = true)\n", " |-- trip_distance: float (nullable = true)\n", " |-- payment_type: integer (nullable = true)\n", " |-- total_amount: float (nullable = true)\n", "\n" ] } ], "source": [ "df_rides.printSchema()" ] }, { "cell_type": "code", "execution_count": null, "id": "f1cdb53e-f477-4137-8412-6915d7772125", "metadata": { "tags": [] }, "outputs": [], "source": [ "df_rides.show()" ] }, { "cell_type": "markdown", "id": "60277fdc-2797-4b23-9ecf-956b76db5778", "metadata": { "tags": [] }, "source": [ "### 2 Sink Operation & Streaming Query\n", "\n", "through `writeStream`\n", "\n", "---\n", "**Output Sinks**\n", "- File Sink: stores the output to the directory\n", "- Kafka Sink: stores the output to one or more topics in Kafka\n", "- Foreach Sink:\n", "- (for debugging) Console Sink, Memory Sink\n", "\n", "Further details can be found in [Output Sinks](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-sinks)\n", "\n", "---\n", "There are three types of **Output Modes**:\n", "- Complete: The whole Result Table will be outputted to the sink after every trigger. This is supported for aggregation queries.\n", "- Append (default): Only new rows are added to the Result Table\n", "- Update: Only updated rows are outputted\n", "\n", "[Output Modes](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes) differs based on the set of transformations applied to the streaming data. \n", "\n", "--- \n", "**Triggers**\n", "\n", "The [trigger settings](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#triggers) of a streaming query define the timing of streaming data processing. Spark streaming support micro-batch streamings schema and you can select following options based on requirements.\n", "\n", "- default-micro-batch-mode\n", "- fixed-interval-micro-batch-mode\n", "- one-time-micro-batch-mode\n", "- available-now-micro-batch-mode\n" ] }, { "cell_type": "markdown", "id": "02ca9b08-aa61-46cd-b946-4457ce2cdf5d", "metadata": { "tags": [] }, "source": [ "#### Console and Memory Sink" ] }, { "cell_type": "code", "execution_count": 11, "id": "74c72469-4c37-417c-a866-a1c1ef75ae8b", "metadata": { "tags": [] }, "outputs": [], "source": [ "def sink_console(df, output_mode: str = 'complete', processing_time: str = '5 seconds'):\n", " write_query = df.writeStream \\\n", " .outputMode(output_mode) \\\n", " .trigger(processingTime=processing_time) \\\n", " .format(\"console\") \\\n", " .option(\"truncate\", False) \\\n", " .start()\n", " return write_query # pyspark.sql.streaming.StreamingQuery" ] }, { "cell_type": "code", "execution_count": 12, "id": "d866c7ba-f8e9-475d-830a-50ffb2c5472b", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "24/03/11 00:30:31 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-2b8e8845-1369-4653-8c23-c45a98e194a9. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n", "24/03/11 00:30:31 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "24/03/11 00:30:32 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n", "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n", "24/03/11 00:30:32 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n", "24/03/11 00:30:33 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "-------------------------------------------\n", "Batch: 0\n", "-------------------------------------------\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|1 |2020-07-01 00:25:32 |2020-07-01 00:33:39 |1 |1.5 |2 |9.3 |\n", "|1 |2020-07-01 00:03:19 |2020-07-01 00:25:43 |1 |9.5 |1 |27.8 |\n", "|2 |2020-07-01 00:15:11 |2020-07-01 00:29:24 |1 |5.85 |2 |22.3 |\n", "|2 |2020-07-01 00:30:49 |2020-07-01 00:38:26 |1 |1.9 |1 |14.16 |\n", "|2 |2020-07-01 00:31:26 |2020-07-01 00:38:02 |1 |1.25 |2 |7.8 |\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "\n" ] } ], "source": [ "write_query = sink_console(df_rides, output_mode='append')" ] }, { "cell_type": "code", "execution_count": 13, "id": "a9bfa73f-a8cc-4988-a8cf-bf31ee6c449c", "metadata": { "tags": [] }, "outputs": [], "source": [ "def sink_memory(df, query_name, query_template):\n", " write_query = df \\\n", " .writeStream \\\n", " .queryName(query_name) \\\n", " .format('memory') \\\n", " .start()\n", " query_str = query_template.format(table_name=query_name)\n", " query_results = spark.sql(query_str)\n", " return write_query, query_results" ] }, { "cell_type": "code", "execution_count": 14, "id": "b31d0b76-e917-44e7-a14d-f9ce6901c23a", "metadata": { "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "24/03/11 00:30:42 WARN ResolveWriteToStream: Temporary checkpoint location created which is deleted normally when the query didn't fail: /tmp/temporary-c7621425-b7fb-47fe-8b42-791c9c5d3186. If it's required to delete it under any circumstances, please set spark.sql.streaming.forceDeleteTempCheckpointLocation to true. Important to know deleting temp checkpoint folder is best effort.\n", "24/03/11 00:30:42 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.\n", "24/03/11 00:30:43 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n", "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n", "24/03/11 00:30:43 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "24/03/11 00:30:43 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", " \r" ] } ], "source": [ "query_name = 'vendor_id_counts'\n", "query_template = 'select count(distinct(vendor_id)) from {table_name}'\n", "write_query, df_vendor_id_counts = sink_memory(df=df_rides, query_name=query_name, query_template=query_template)" ] }, { "cell_type": "code", "execution_count": 15, "id": "4ba56111-83bf-4028-ac65-565e0190f310", "metadata": { "tags": [] }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n" ] }, { "data": { "text/plain": [ "{'message': 'Waiting for data to arrive',\n", " 'isDataAvailable': False,\n", " 'isTriggerActive': False}" ] }, "execution_count": 15, "metadata": {}, "output_type": "execute_result" }, { "name": "stdout", "output_type": "stream", "text": [ "-------------------------------------------\n", "Batch: 1\n", "-------------------------------------------\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|1 |2020-07-01 00:25:32 |2020-07-01 00:33:39 |1 |1.5 |2 |9.3 |\n", "|1 |2020-07-01 00:03:19 |2020-07-01 00:25:43 |1 |9.5 |1 |27.8 |\n", "|2 |2020-07-01 00:15:11 |2020-07-01 00:29:24 |1 |5.85 |2 |22.3 |\n", "|2 |2020-07-01 00:30:49 |2020-07-01 00:38:26 |1 |1.9 |1 |14.16 |\n", "|2 |2020-07-01 00:31:26 |2020-07-01 00:38:02 |1 |1.25 |2 |7.8 |\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "\n", "-------------------------------------------\n", "Batch: 2\n", "-------------------------------------------\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|vendor_id|tpep_pickup_datetime|tpep_dropoff_datetime|passenger_count|trip_distance|payment_type|total_amount|\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "|1 |2020-07-01 00:25:32 |2020-07-01 00:33:39 |1 |1.5 |2 |9.3 |\n", "|1 |2020-07-01 00:03:19 |2020-07-01 00:25:43 |1 |9.5 |1 |27.8 |\n", "|2 |2020-07-01 00:15:11 |2020-07-01 00:29:24 |1 |5.85 |2 |22.3 |\n", "|2 |2020-07-01 00:30:49 |2020-07-01 00:38:26 |1 |1.9 |1 |14.16 |\n", "|2 |2020-07-01 00:31:26 |2020-07-01 00:38:02 |1 |1.25 |2 |7.8 |\n", "+---------+--------------------+---------------------+---------------+-------------+------------+------------+\n", "\n" ] } ], "source": [ "print(type(write_query)) # pyspark.sql.streaming.StreamingQuery\n", "write_query.status" ] }, { "cell_type": "code", "execution_count": 18, "id": "7cc37bda-9cfa-402b-9d42-a6ba5271476b", "metadata": { "tags": [] }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------------------------+\n", "|count(DISTINCT vendor_id)|\n", "+-------------------------+\n", "| 2|\n", "+-------------------------+\n", "\n" ] } ], "source": [ "df_vendor_id_counts.show()" ] }, { "cell_type": "code", "execution_count": 19, "id": "88862ca9-4d89-487e-987f-08a2b9e83efe", "metadata": { "tags": [] }, "outputs": [], "source": [ "write_query.stop()" ] }, { "cell_type": "markdown", "id": "443d4041-06db-4a4a-89c1-348848cc7ca8", "metadata": { "tags": [] }, "source": [ "#### Kafka Sink\n", "\n", "To write stream results to `kafka-topic`, the stream dataframe has at least a column with name `value`.\n", "\n", "Therefore before starting `writeStream` in kafka format, dataframe needs to be updated accordingly.\n", "\n", "More information regarding kafka sink expected data structure [here](https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#writing-data-to-kafka)\n" ] }, { "cell_type": "code", "execution_count": 20, "id": "8b08a013-d039-41cf-94fd-a1a57571d25f", "metadata": { "scrolled": true, "tags": [] }, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:36 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:37 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:39 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:40 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:41 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:42 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:43 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:44 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:45 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:46 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:47 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:48 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:49 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:50 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:51 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:52 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:53 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:54 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:55 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:56 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:57 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:34:58 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:00 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:01 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:02 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:03 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:04 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:05 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:06 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:07 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:08 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:09 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:10 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:11 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:12 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:13 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:14 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:16 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:17 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:17 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:19 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:20 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:21 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:22 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:23 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:24 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:25 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:26 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:27 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:28 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:29 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:30 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:31 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:32 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:33 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:34 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:35 WARN KafkaOffsetReaderAdmin: Error in attempt 1 getting Kafka offsets: \n", "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088535000, tries=1, nextAllowedTryMs=1710088535101) timed out at 1710088535001 after 1 attempt(s)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n", "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n", "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n", "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n", "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n", "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n", "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n", "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n", "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n", "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n", "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n", "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n", "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n", "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088535000, tries=1, nextAllowedTryMs=1710088535101) timed out at 1710088535001 after 1 attempt(s)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n", "24/03/11 00:35:35 WARN NetworkClient: [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:36 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n", "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n", "24/03/11 00:35:36 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n", "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:37 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:37 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:39 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:40 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:41 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:42 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:43 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:44 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:45 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:46 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:47 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:48 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:49 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:50 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:51 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:52 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:53 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:55 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:55 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:57 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:58 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:35:59 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:00 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:01 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:02 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:03 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:04 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:06 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:07 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:08 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:09 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:10 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:11 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:12 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:13 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:14 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:15 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:16 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:17 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:18 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:19 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:20 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:22 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:23 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:24 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:25 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:26 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:27 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:28 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:29 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:30 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:31 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:32 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:33 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:35 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:35 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:36 WARN KafkaOffsetReaderAdmin: Error in attempt 2 getting Kafka offsets: \n", "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088596058, tries=1, nextAllowedTryMs=1710088596159) timed out at 1710088596059 after 1 attempt(s)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n", "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n", "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n", "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n", "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n", "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n", "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n", "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n", "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n", "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n", "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n", "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n", "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n", "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088596058, tries=1, nextAllowedTryMs=1710088596159) timed out at 1710088596059 after 1 attempt(s)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n", "24/03/11 00:36:36 WARN NetworkClient: [AdminClient clientId=adminclient-3] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:37 WARN ClientUtils: Couldn't resolve server broker:29092 from bootstrap.servers as DNS resolution failed for broker\n", "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'key.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'value.deserializer' was supplied but isn't a known config.\n", "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'enable.auto.commit' was supplied but isn't a known config.\n", "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'max.poll.records' was supplied but isn't a known config.\n", "24/03/11 00:36:37 WARN AdminClientConfig: The configuration 'auto.offset.reset' was supplied but isn't a known config.\n", "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:38 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:38 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:40 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:41 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:42 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:43 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:44 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:45 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:46 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:47 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:47 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:48 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:49 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:50 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:52 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:52 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:54 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:55 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:56 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:57 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:58 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:36:59 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:00 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:01 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:02 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:03 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:05 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:05 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:06 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:08 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:09 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:10 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:11 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:12 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:13 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:14 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:15 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:16 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:17 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:18 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:19 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:20 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:21 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:22 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:23 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:24 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:25 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:26 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:27 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:28 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:29 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:31 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:32 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:33 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:34 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:35 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:36 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:37 WARN KafkaOffsetReaderAdmin: Error in attempt 3 getting Kafka offsets: \n", "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n", "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n", "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n", "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n", "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n", "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n", "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n", "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n", "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n", "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n", "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n", "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n", "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n", "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n", "24/03/11 00:37:37 WARN NetworkClient: [AdminClient clientId=adminclient-4] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.\n", "24/03/11 00:37:38 ERROR MicroBatchExecution: Query [id = 4dfba771-eff7-49e7-a3ff-f1aa03a6e840, runId = 0f86ad02-1d50-487a-97c7-72790d8857d8] terminated with error\n", "java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)\n", "\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions(ConsumerStrategy.scala:66)\n", "\tat org.apache.spark.sql.kafka010.ConsumerStrategy.retrieveAllPartitions$(ConsumerStrategy.scala:65)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.retrieveAllPartitions(ConsumerStrategy.scala:102)\n", "\tat org.apache.spark.sql.kafka010.SubscribeStrategy.assignedTopicPartitions(ConsumerStrategy.scala:113)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.$anonfun$partitionsAssignedToAdmin$1(KafkaOffsetReaderAdmin.scala:499)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.withRetries(KafkaOffsetReaderAdmin.scala:518)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.partitionsAssignedToAdmin(KafkaOffsetReaderAdmin.scala:498)\n", "\tat org.apache.spark.sql.kafka010.KafkaOffsetReaderAdmin.fetchLatestOffsets(KafkaOffsetReaderAdmin.scala:297)\n", "\tat org.apache.spark.sql.kafka010.KafkaMicroBatchStream.latestOffset(KafkaMicroBatchStream.scala:132)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$4(MicroBatchExecution.scala:491)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$2(MicroBatchExecution.scala:490)\n", "\tat scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)\n", "\tat scala.collection.Iterator.foreach(Iterator.scala:943)\n", "\tat scala.collection.Iterator.foreach$(Iterator.scala:943)\n", "\tat scala.collection.AbstractIterator.foreach(Iterator.scala:1431)\n", "\tat scala.collection.IterableLike.foreach(IterableLike.scala:74)\n", "\tat scala.collection.IterableLike.foreach$(IterableLike.scala:73)\n", "\tat scala.collection.AbstractIterable.foreach(Iterable.scala:56)\n", "\tat scala.collection.TraversableLike.map(TraversableLike.scala:286)\n", "\tat scala.collection.TraversableLike.map$(TraversableLike.scala:279)\n", "\tat scala.collection.AbstractTraversable.map(Traversable.scala:108)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$constructNextBatch$1(MicroBatchExecution.scala:479)\n", "\tat scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.withProgressLocked(MicroBatchExecution.scala:810)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.constructNextBatch(MicroBatchExecution.scala:475)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$2(MicroBatchExecution.scala:268)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken(ProgressReporter.scala:427)\n", "\tat org.apache.spark.sql.execution.streaming.ProgressReporter.reportTimeTaken$(ProgressReporter.scala:425)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.$anonfun$runActivatedStream$1(MicroBatchExecution.scala:249)\n", "\tat org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:67)\n", "\tat org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:239)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:311)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:900)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:289)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.$anonfun$run$1(StreamExecution.scala:211)\n", "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n", "\tat org.apache.spark.JobArtifactSet$.withActiveJobArtifactState(JobArtifactSet.scala:94)\n", "\tat org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:211)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=describeTopics, deadlineMs=1710088657106, tries=1, nextAllowedTryMs=1710088657207) timed out at 1710088657107 after 1 attempt(s)\n", "Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: describeTopics\n" ] } ], "source": [ "def prepare_dataframe_to_kafka_sink(df, value_columns, key_column=None):\n", " columns = df.columns\n", " df = df.withColumn(\"value\", F.concat_ws(', ',*value_columns)) \n", " if key_column:\n", " df = df.withColumnRenamed(key_column,\"key\")\n", " df = df.withColumn(\"key\",df.key.cast('string'))\n", " return df.select(['key', 'value'])\n", " \n", "def sink_kafka(df, topic, output_mode='append'):\n", " write_query = df.writeStream \\\n", " .format(\"kafka\") \\\n", " .option(\"kafka.bootstrap.servers\", \"localhost:9092,broker:29092\") \\\n", " .outputMode(output_mode) \\\n", " .option(\"topic\", topic) \\\n", " .option(\"checkpointLocation\", \"checkpoint\") \\\n", " .start()\n", " return write_query" ] }, { "cell_type": "markdown", "id": "e4cb2140-9f2e-4914-b74c-be4c18cdbe8a", "metadata": {}, "source": [] }, { "cell_type": "code", "execution_count": null, "id": "63abe115-879c-4863-97d3-b22cda7f7469", "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.8" } }, "nbformat": 4, "nbformat_minor": 5 }