2_spark_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "36a2acf1",
   "metadata": {},
   "source": [
    "<center><h1>Management and Analysis of Physics Dataset (MOD. B) </h1></center>\n",
    "<center><h2> Project 5 - Streaming processing of cosmic rays using Drift Tubes detectors</h2></center>\n",
    "<center><h2>Group 2305</h2></center>\n",
    "\n",
    "<center><style>\n",
    "    table {font-size: 24px;}\n",
    "</style></center>\n",
    "\n",
    "| Last Name        | First Name         |Student ID|\n",
    "|:----------------:|:------------------:|:--------------:|\n",
    "| Bertinelli       | Gabriele           |1219907 (tri)   |\n",
    "| Bhatti           | Roben              |2091187         |\n",
    "| Bonato           | Diego              |2091250         |\n",
    "| Cacciola         | Martina            |2097476         |\n",
    "\n",
    "<left><h2> Part 2 - Data processing</h2></left>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "23cc2a5d",
   "metadata": {},
   "source": [
    "### Import packages and modules"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "af7f41da",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import json\n",
    "import boto3\n",
    "import json\n",
    "import time\n",
    "\n",
    "from tqdm        import tqdm\n",
    "\n",
    "import kafka\n",
    "from kafka       import KafkaProducer\n",
    "from kafka.admin import KafkaAdminClient, NewTopic\n",
    "\n",
    "\n",
    "\n",
    "from pyspark.sql.functions import from_json, col, countDistinct, count, when, collect_list\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from pyspark.sql.types import StructField, StructType, StringType, DoubleType, IntegerType\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7aab7365",
   "metadata": {},
   "source": [
    "### Session and Spark Context creation\n",
    "\n",
    "With the following command we are asking to the master (and the resource manager) to create an application (Session) with the required resources and configurations.\n",
    "In order to test the performance of the network, we varied the following parameters: \n",
    "- `spark.executor.instances`: controls the number of executors requested (Excecutors perform the actual computations on the data. They are responsible for executing these tasks in parallel and returning the results back to the driver program).\n",
    "- `spark.executor.cores`: specifies the number of CPU cores that are allocated to each Executor\n",
    "- `spark.sql.shuffle.partitions`: configures the number of partitions that are used when shuffling data for joins or aggregations.\n",
    "- `spark.sql.execution.arrow.pyspark.enabled`: Apache Arrow is an in-memory columnar data format that is used in Spark to efficiently transfer data between JVM and Python processes. Arrow is available as an optimization when converting a Spark DataFrame to a Pandas DataFrame and when creating a Spark DataFrame from a Pandas DataFrame.\n",
    "\n",
    "The Spark Context is the driver application program used to submit applications to Spark, and it is used to work with RDDs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "0d27936c",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/home/bertinelli/spark/spark/lib/python3.8/site-packages/pyspark/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/bertinelli/.ivy2/cache\n",
      "The jars for the packages stored in: /home/bertinelli/.ivy2/jars\n",
      "org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-9f51bc1d-490e-477c-9f70-41543341391b;1.0\n",
      "\tconfs: [default]\n",
      "\tfound org.apache.spark#spark-sql-kafka-0-10_2.12;3.3.2 in central\n",
      "\tfound org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.2 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",
      ":: resolution report :: resolve 757ms :: artifacts dl 65ms\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-sql-kafka-0-10_2.12;3.3.2 from central in [default]\n",
      "\torg.apache.spark#spark-token-provider-kafka-0-10_2.12;3.3.2 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.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     |   12  |   0   |   0   |   0   ||   12  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-9f51bc1d-490e-477c-9f70-41543341391b\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 12 already retrieved (0kB/17ms)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/07/09 18:53:32 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .master(\"##\")\\\n",
    "    .appName(\"cosmic_rays_spark\")\\\n",
    "    .config(\"spark.executor.instances\", 10)\\\n",
    "    .config(\"spark.executor.cores\",1)\\\n",
    "    .config(\"spark.sql.shuffle.partitions\", 10)\\\n",
    "    .config(\"spark.executor.memory\", \"1500m\")\\\n",
    "    .config(\"spark.driver.memory\", \"1g\")\\\n",
    "    .config(\"spark.sql.execution.arrow.pyspark.enabled\", \"true\")\\\n",
    "    .config(\"spark.sql.execution.arrow.pyspark.fallback.enabled\", \"true\")\\\n",
    "    .config(\"spark.sql.adaptive.enabled\", \"false\")\\\n",
    "    .config(\"spark.sql.adaptive.coalescePartitions.enabled\", \"false\")\\\n",
    "    .config(\"spark.jars.packages\",\"org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.2\")\\\n",
    "    .config(\"spark.sql.streaming.forceDeleteTempCheckpointLocation\", \"true\")\\\n",
    "    .config('spark.eventLog.enabled', 'true')\\\n",
    "    .config('spark.sql.streaming.stateStore.stateSchemaCheck', 'false')\\\n",
    "    .config(\"spark.sql.streaming.numRecentProgressUpdates\", 1000)\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "808a4c4c",
   "metadata": {
    "vscode": {
     "languageId": "plaintext"
    }
   },
   "outputs": [],
   "source": [
    "sc = spark.sparkContext\n",
    "sc"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "33aed84b",
   "metadata": {},
   "source": [
    "### Producer creation\n",
    "Here we create the producer, which is then used to send the message containing the “cleaned” data to the dashboard via producer.send (in the function) and foreachBatch (applying the function)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ae4db05a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the list of brokers in the cluster\n",
    "KAFKA_BOOTSTRAP_SERVER = '##' #kafka broker"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "10834a24",
   "metadata": {},
   "outputs": [],
   "source": [
    "# producer definition\n",
    "producer = KafkaProducer(bootstrap_servers = KAFKA_BOOTSTRAP_SERVER,\n",
    "                         batch_size=16000, #16MB\n",
    "                          linger_ms=20  ) #ms\n",
    "\n",
    "        \n",
    "# KAFKA ADMIN is responsible for creating/deleting topics\n",
    "\n",
    "# connecting to client \n",
    "kafka_admin = KafkaAdminClient(bootstrap_servers = KAFKA_BOOTSTRAP_SERVER)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "53d9c7e2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['data_clean', 'test_clean_1', 'data_raw', '__consumer_offsets']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "kafka_admin.list_topics()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "00057b81",
   "metadata": {},
   "source": [
    "### Data preprocessing\n",
    "\n",
    "We create a DataFrame representing the stream of input lines from Kafka by connecting to the appropriate servers and topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7b2b7827",
   "metadata": {},
   "outputs": [],
   "source": [
    "# read streaming df from kafka\n",
    "inputDF = spark\\\n",
    "    .readStream\\\n",
    "    .format(\"kafka\")\\\n",
    "    .option(\"kafka.bootstrap.servers\", KAFKA_BOOTSTRAP_SERVER)\\\n",
    "    .option(\"kafkaConsumer.pollTimeoutMs\", 1000)\\\n",
    "    .option('subscribe', 'data_raw')\\\n",
    "    .option(\"startingOffsets\", \"latest\") \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "081c2b3f",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "inputDF.isStreaming"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5135360e",
   "metadata": {},
   "source": [
    "We extract the values from the Kafka message, and we use `selectExpr` to create a table with the desired columns. \n",
    "`selectExpr()` is a function that takes a set of SQL expressions in a string to execute. This gives the ability to run SQL like expressions without creating a temporary table and views. `selectExpr()` just has one signature that takes SQL expression in a String and returns a new DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "0461a7d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# extract the value from the kafka message\n",
    "rawraw_data = inputDF.select(col(\"value\").cast(\"string\")).alias(\"csv\")\n",
    "\n",
    "# split the csv line in the corresponding fields\n",
    "raw_data = rawraw_data.selectExpr(\"cast(split(value, ',')[0] as int) as HEAD\",\n",
    "                                   \"cast(split(value, ',')[1] as int) as FPGA\",\n",
    "                                   \"cast(split(value, ',')[2] as int) as TDC_CHANNEL\",\n",
    "                                   \"cast(split(value, ',')[3] as long) as ORBIT_CNT\",\n",
    "                                   \"cast(split(value, ',')[4] as int) as BX_COUNTER\",\n",
    "                                   \"cast(split(value, ',')[5] as double) as TDC_MEAS\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "f298698f",
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "#raw_data.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "917be55e",
   "metadata": {},
   "source": [
    "We select only the useful data (`HEAD=2`) and we apply a mapping between the data-format and the detectors, creating a new column `CHAMBER`. We eventually remove all rows with null values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "c0913dce",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# selecting the wanted column\n",
    "raw_data = raw_data.filter(raw_data.HEAD == 2)\n",
    "\n",
    "# add CHAMBER column\n",
    "raw_data = raw_data.withColumn(\"CHAMBER\", \\\n",
    "                    when((raw_data.FPGA == 0) & (raw_data.TDC_CHANNEL>=0) & (raw_data.TDC_CHANNEL<64), 0) \\\n",
    "                   .when((raw_data.FPGA == 0) & (raw_data.TDC_CHANNEL>=64) & (raw_data.TDC_CHANNEL<128), 1 ) \\\n",
    "                   .when((raw_data.FPGA == 1) & (raw_data.TDC_CHANNEL>=0) & (raw_data.TDC_CHANNEL<64), 2 ) \\\n",
    "                   .when((raw_data.FPGA == 1) & (raw_data.TDC_CHANNEL>=64) & (raw_data.TDC_CHANNEL<128), 3 )\\\n",
    "                  ).na.drop()\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "97dbf570",
   "metadata": {},
   "source": [
    "### Data processing\n",
    "\n",
    "We create a function that will be applied to each batch of data in order to compute the following piece of data:\n",
    "1. total number of processed hits, post-clensing (1 value per batch) &rarr; `hit_count`\n",
    "2. total number of processed hits, post-clensing, per chamber (4 values per batch) &rarr; `hit_count_chamber`\n",
    "3. histogram of the counts of active `TDC_CHANNEL`, per chamber (4 arrays per batch) &rarr; `ch*_tdc_counts_list`\n",
    "4. histogram of the total number of active `TDC_CHANNEL` in each `ORBIT_CNT`, per chamber (4 arrays per batch) &rarr; `ch*_orbit_counts_list`\n",
    "\n",
    "Finally, we create a json message that will be sent to a Kafka topic via `producer.send()`.\n",
    "Additionally, we print the time taken by each batch from the beginning of the data processing phase until the message is sent to the topic.\n",
    "\n",
    "We use the `.persist()` function on the initial and intermediate dataframes for tasks 3 and 4.\n",
    "It is a good practice to persist the DataFrame in memory before performing operations on it repeatedly. This helps in caching the DataFrame's partitions in memory and avoids unnecessary re-computation when the DataFrame is accessed multiple times."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "3077ac84",
   "metadata": {},
   "outputs": [],
   "source": [
    "from numpyencoder import NumpyEncoder\n",
    "\n",
    "ID = -1\n",
    "\n",
    "# function to apply to each batch: writes and sends a kafka message at the end\n",
    "def batch_processing(df, epoch_id):\n",
    "\n",
    "    df = df.persist()\n",
    "\n",
    "    # 1: total number of processed hits, post-cleansing (1 value per batch)\n",
    "    \n",
    "    start = time.time()\n",
    "    \n",
    "    hit_count = df.count()\n",
    "    \n",
    "    \n",
    "    # 2: total number of processed hits, post-cleansing, per chamber (4 values per batch)\n",
    "\n",
    "\n",
    "    hit_count_chamber = df.groupby('CHAMBER').agg(count('TDC_CHANNEL').alias('HIT_COUNT'))\\\n",
    "                        .sort(\"CHAMBER\").select('HIT_COUNT')\\\n",
    "                        .agg(collect_list('HIT_COUNT')).collect()\n",
    "\n",
    "\n",
    "\n",
    "    # 3: histogram of the counts of active TDC_CHANNEL, per chamber (4 arrays per batch)\n",
    "\n",
    "    tdc_counts = df.groupby(['CHAMBER','TDC_CHANNEL']).agg(count('TDC_CHANNEL').alias('TDC_COUNTS'))\n",
    "    tdc_counts = tdc_counts.persist()\n",
    "\n",
    "    # Filter the tdc_counts DataFrame for each chamber \n",
    "    \n",
    "    ch0_tdc_counts = tdc_counts.filter(tdc_counts.CHAMBER == 0).select('TDC_CHANNEL','TDC_COUNTS')\\\n",
    "                    .sort(\"TDC_CHANNEL\").toPandas()\n",
    "    \n",
    "    ch1_tdc_counts = tdc_counts.filter(tdc_counts.CHAMBER == 1).select('TDC_CHANNEL','TDC_COUNTS')\\\n",
    "                    .sort(\"TDC_CHANNEL\").toPandas()\n",
    "    \n",
    "    ch2_tdc_counts = tdc_counts.filter(tdc_counts.CHAMBER == 2).select('TDC_CHANNEL','TDC_COUNTS')\\\n",
    "                    .sort(\"TDC_CHANNEL\").toPandas()\n",
    "    \n",
    "    ch3_tdc_counts = tdc_counts.filter(tdc_counts.CHAMBER == 3).select('TDC_CHANNEL','TDC_COUNTS')\\\n",
    "                    .sort(\"TDC_CHANNEL\").toPandas()\n",
    "\n",
    "    \n",
    "    \n",
    "    #Save it in a list\n",
    "    \n",
    "    ch0_tdc_channels_list = list(ch0_tdc_counts['TDC_CHANNEL'])\n",
    "    ch0_tdc_counts_list   = list(ch0_tdc_counts['TDC_COUNTS'])\n",
    "\n",
    "    ch1_tdc_channels_list = list(ch1_tdc_counts['TDC_CHANNEL'])\n",
    "    ch1_tdc_counts_list   = list(ch1_tdc_counts['TDC_COUNTS'])\n",
    "    \n",
    "    ch2_tdc_channels_list = list(ch2_tdc_counts['TDC_CHANNEL'])\n",
    "    ch2_tdc_counts_list   = list(ch2_tdc_counts['TDC_COUNTS'])\n",
    "    \n",
    "    ch3_tdc_channels_list = list(ch3_tdc_counts['TDC_CHANNEL'])\n",
    "    ch3_tdc_counts_list   = list(ch3_tdc_counts['TDC_COUNTS'])\n",
    "    \n",
    "    \n",
    "\n",
    "    # 4: histogram of the total number of active TDC_CHANNEL in each ORBIT_CNT, per chamber (4 arrays per batch)\n",
    "\n",
    "    orbit_count=df.groupby(['CHAMBER','ORBIT_CNT']).agg(countDistinct(\"TDC_CHANNEL\").alias('TDC_ORBIT'))\n",
    "    orbit_count = orbit_count.persist()\n",
    "\n",
    "    ch0_orbit_counts = orbit_count.filter(orbit_count.CHAMBER == 0).select('ORBIT_CNT','TDC_ORBIT')\\\n",
    "                    .sort(\"ORBIT_CNT\").toPandas()\n",
    "    \n",
    "    ch1_orbit_counts = orbit_count.filter(orbit_count.CHAMBER == 1).select('ORBIT_CNT','TDC_ORBIT')\\\n",
    "                    .sort(\"ORBIT_CNT\").toPandas()\n",
    "    \n",
    "    ch2_orbit_counts = orbit_count.filter(orbit_count.CHAMBER == 2).select('ORBIT_CNT','TDC_ORBIT')\\\n",
    "                    .sort(\"ORBIT_CNT\").toPandas()\n",
    "    \n",
    "    ch3_orbit_counts = orbit_count.filter(orbit_count.CHAMBER == 3).select('ORBIT_CNT','TDC_ORBIT')\\\n",
    "                    .sort(\"ORBIT_CNT\").toPandas()\n",
    "    \n",
    "    #Save it in a list\n",
    "    \n",
    "    ch0_orbit_list          = list(ch0_orbit_counts['ORBIT_CNT'])\n",
    "    ch0_orbit_counts_list   = list(ch0_orbit_counts['TDC_ORBIT'])\n",
    "\n",
    "    ch1_orbit_list          = list(ch1_orbit_counts['ORBIT_CNT'])\n",
    "    ch1_orbit_counts_list   = list(ch1_orbit_counts['TDC_ORBIT'])\n",
    "\n",
    "    ch2_orbit_list          = list(ch2_orbit_counts['ORBIT_CNT'])\n",
    "    ch2_orbit_counts_list   = list(ch2_orbit_counts['TDC_ORBIT'])\n",
    "    \n",
    "    ch3_orbit_list          = list(ch3_orbit_counts['ORBIT_CNT'])\n",
    "    ch3_orbit_counts_list   = list(ch3_orbit_counts['TDC_ORBIT'])\n",
    "    \n",
    "\n",
    " \n",
    "    \n",
    "    df.unpersist()\n",
    "    tdc_counts.unpersist()\n",
    "    orbit_count.unpersist()\n",
    "        \n",
    "    \n",
    "    global ID\n",
    "    ID += 1\n",
    "\n",
    "    # prepare message to send to kafka\n",
    "    \n",
    "    msg = {\n",
    "\n",
    "        'msg_ID': ID,\n",
    "        'hit_count': hit_count,\n",
    "        'hit_count_chamber': hit_count_chamber[0][0],\n",
    "           \n",
    "        'tdc_counts_chamber': {\n",
    "            '0': {\n",
    "                'bin_edges': ch0_tdc_channels_list,\n",
    "                'hist_counts': ch0_tdc_counts_list\n",
    "            },\n",
    "            '1': {\n",
    "                'bin_edges': ch1_tdc_channels_list,\n",
    "                'hist_counts': ch1_tdc_counts_list\n",
    "            },\n",
    "            '2': {\n",
    "                'bin_edges': ch2_tdc_channels_list,\n",
    "                'hist_counts': ch2_tdc_counts_list\n",
    "            },\n",
    "            '3': {\n",
    "                'bin_edges': ch3_tdc_channels_list,\n",
    "                'hist_counts': ch3_tdc_counts_list\n",
    "            }\n",
    "        },\n",
    "        'active_tdc_chamber': {\n",
    "            '0': {\n",
    "                'bin_edges': ch0_orbit_list,\n",
    "                'hist_counts': ch0_orbit_counts_list\n",
    "            },\n",
    "            '1': {\n",
    "                'bin_edges': ch1_orbit_list,\n",
    "                'hist_counts': ch1_orbit_counts_list\n",
    "            },\n",
    "            '2': {\n",
    "                'bin_edges': ch2_orbit_list,\n",
    "                'hist_counts': ch2_orbit_counts_list\n",
    "            },\n",
    "            '3': {\n",
    "                'bin_edges': ch3_orbit_list,\n",
    "                'hist_counts': ch3_orbit_counts_list\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "   \n",
    "\n",
    "    \n",
    "    \n",
    "    \n",
    "    producer.send('data_clean', json.dumps(msg).encode('utf-8'))\n",
    "    producer.flush()\n",
    "    #producer.poll(0)\n",
    "    \n",
    "    stop = time.time() - start\n",
    "    \n",
    "    print(f'\\nTime: {stop}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a4bb6b58",
   "metadata": {},
   "source": [
    "### Sending message to Kafka topic\n",
    "\n",
    "Here, we apply the function defined earlier, `batch_processing()`, to each batch in the streaming dataset `raw_data`, through the `foreachBatch()` function.\n",
    "`.trigger()` is a method that allows you to define the trigger for a streaming query. A trigger determines when the streaming query should be executed and how often it should process the data. To investigate performance differences, we varies the `processingTime` parameter."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "ba8eb0a1",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 1:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "False"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 23.333239793777466\n",
      "23/07/09 18:54:13 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 26411 milliseconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 56:=====>                                                   (1 + 8) / 10]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/07/09 18:54:16 WARN TaskSetManager: Lost task 9.0 in stage 56.0 (TID 210) (10.67.22.77 executor 0): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3576 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n",
      "23/07/09 18:54:16 WARN TaskSetManager: Lost task 4.0 in stage 56.0 (TID 205) (10.67.22.77 executor 2): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3611 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n",
      "23/07/09 18:54:17 WARN TaskSetManager: Lost task 1.0 in stage 56.0 (TID 202) (10.67.22.111 executor 5): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3619 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n",
      "23/07/09 18:54:17 WARN TaskSetManager: Lost task 0.0 in stage 56.0 (TID 201) (10.67.22.77 executor 1): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3684 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n",
      "23/07/09 18:54:17 WARN TaskSetManager: Lost task 3.0 in stage 56.0 (TID 204) (10.67.22.111 executor 4): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3664 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n",
      "23/07/09 18:54:17 WARN TaskSetManager: Lost task 6.0 in stage 56.0 (TID 207) (10.67.22.111 executor 7): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3572 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 56:=================>                                       (3 + 7) / 10]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23/07/09 18:54:17 WARN TaskSetManager: Lost task 2.0 in stage 56.0 (TID 203) (10.67.22.111 executor 6): java.util.concurrent.TimeoutException: Cannot fetch record for offset 3468 in 1000 milliseconds\n",
      "\tat org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.fetch(KafkaDataConsumer.scala:97)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchData(KafkaDataConsumer.scala:552)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.fetchRecord(KafkaDataConsumer.scala:476)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:313)\n",
      "\tat org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:618)\n",
      "\tat org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:290)\n",
      "\tat org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:64)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)\n",
      "\tat scala.Option.exists(Option.scala:376)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.advanceToNextIter(DataSourceRDD.scala:97)\n",
      "\tat org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)\n",
      "\tat org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)\n",
      "\tat org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)\n",
      "\tat org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:760)\n",
      "\tat org.apache.spark.sql.execution.columnar.DefaultCachedBatchSerializer$$anon$1.hasNext(InMemoryRelation.scala:118)\n",
      "\tat scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIterator(MemoryStore.scala:223)\n",
      "\tat org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:302)\n",
      "\tat org.apache.spark.storage.BlockManager.$anonfun$doPutIterator$1(BlockManager.scala:1523)\n",
      "\tat org.apache.spark.storage.BlockManager.org$apache$spark$storage$BlockManager$$doPut(BlockManager.scala:1450)\n",
      "\tat org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1514)\n",
      "\tat org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:1337)\n",
      "\tat org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:376)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:327)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)\n",
      "\tat org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:365)\n",
      "\tat org.apache.spark.rdd.RDD.iterator(RDD.scala:329)\n",
      "\tat org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)\n",
      "\tat org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)\n",
      "\tat org.apache.spark.scheduler.Task.run(Task.scala:136)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)\n",
      "\tat org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)\n",
      "\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n",
      "\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n",
      "\tat java.base/java.lang.Thread.run(Thread.java:829)\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 17.259406089782715\n",
      "23/07/09 18:54:30 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 17611 milliseconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 7.772867679595947\n",
      "23/07/09 18:54:39 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 8035 milliseconds\n",
      "\n",
      "Time: 5.393748760223389\n",
      "23/07/09 18:54:44 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 5635 milliseconds\n",
      "\n",
      "Time: 5.390012741088867\n",
      "23/07/09 18:54:50 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 5624 milliseconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 5.221779108047485\n",
      "23/07/09 18:54:55 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 5494 milliseconds\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 4.829960346221924\n",
      "23/07/09 18:55:00 WARN ProcessingTimeExecutor: Current batch is falling behind. The trigger interval is 5000 milliseconds, but spent 5061 milliseconds\n",
      "\n",
      "Time: 4.542272090911865\n",
      "\n",
      "Time: 4.629445314407349\n",
      "\n",
      "Time: 4.434342861175537\n",
      "\n",
      "Time: 3.8700287342071533\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.9892547130584717\n",
      "\n",
      "Time: 3.41048264503479\n",
      "\n",
      "Time: 3.506680727005005\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.6059868335723877\n",
      "\n",
      "Time: 3.1846048831939697\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.816849708557129\n",
      "\n",
      "Time: 3.565668821334839\n",
      "\n",
      "Time: 3.7120273113250732\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.7680063247680664\n",
      "\n",
      "Time: 3.149793863296509\n",
      "\n",
      "Time: 3.757244110107422\n",
      "\n",
      "Time: 2.896088123321533\n",
      "\n",
      "Time: 3.446298122406006\n",
      "\n",
      "Time: 3.2694919109344482\n",
      "\n",
      "Time: 3.2564494609832764\n",
      "\n",
      "Time: 3.259594678878784\n",
      "\n",
      "Time: 2.812570095062256\n",
      "\n",
      "Time: 3.298490285873413\n",
      "\n",
      "Time: 2.913853645324707\n",
      "\n",
      "Time: 3.0727643966674805\n",
      "\n",
      "Time: 3.1094141006469727\n",
      "\n",
      "Time: 2.691243886947632\n",
      "\n",
      "Time: 3.064256429672241\n",
      "\n",
      "Time: 2.956057071685791\n",
      "\n",
      "Time: 3.2700514793395996\n",
      "\n",
      "Time: 2.7510828971862793\n",
      "\n",
      "Time: 3.21443772315979\n",
      "\n",
      "Time: 3.1759955883026123\n",
      "\n",
      "Time: 2.9388937950134277\n",
      "\n",
      "Time: 3.2907140254974365\n",
      "\n",
      "Time: 2.747364044189453\n",
      "\n",
      "Time: 2.9534990787506104\n",
      "\n",
      "Time: 2.66043758392334\n",
      "\n",
      "Time: 3.0427260398864746\n",
      "\n",
      "Time: 3.160946846008301\n",
      "\n",
      "Time: 2.81296443939209\n",
      "\n",
      "Time: 3.1592609882354736\n",
      "\n",
      "Time: 2.575378656387329\n",
      "\n",
      "Time: 2.9755373001098633\n",
      "\n",
      "Time: 3.009711980819702\n",
      "\n",
      "Time: 2.8884880542755127\n",
      "\n",
      "Time: 3.008955478668213\n",
      "\n",
      "Time: 2.801079034805298\n",
      "\n",
      "Time: 2.953573703765869\n",
      "\n",
      "Time: 3.042083501815796\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.994786500930786\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.8760201930999756\n",
      "\n",
      "Time: 3.0294787883758545\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.008552312850952\n",
      "\n",
      "Time: 2.4440202713012695\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.2959132194519043\n",
      "\n",
      "Time: 2.6078410148620605\n",
      "\n",
      "Time: 2.876523017883301\n",
      "\n",
      "Time: 3.189990520477295\n",
      "\n",
      "Time: 3.015429735183716\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.019137382507324\n",
      "\n",
      "Time: 3.0072250366210938\n",
      "\n",
      "Time: 2.7554783821105957\n",
      "\n",
      "Time: 2.823387384414673\n",
      "\n",
      "Time: 2.3266735076904297\n",
      "\n",
      "Time: 2.8080697059631348\n",
      "\n",
      "Time: 2.0749716758728027\n",
      "\n",
      "Time: 2.62727427482605\n",
      "\n",
      "Time: 2.05653715133667\n",
      "\n",
      "Time: 2.4869773387908936\n",
      "\n",
      "Time: 2.8282835483551025\n",
      "\n",
      "Time: 2.3100874423980713\n",
      "\n",
      "Time: 2.729393243789673\n",
      "\n",
      "Time: 2.22770094871521\n",
      "\n",
      "Time: 2.6316006183624268\n",
      "\n",
      "Time: 2.7601494789123535\n",
      "\n",
      "Time: 2.667609214782715\n",
      "\n",
      "Time: 2.6650891304016113\n",
      "\n",
      "Time: 2.3871240615844727\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.52372670173645\n",
      "\n",
      "Time: 2.257821798324585\n",
      "\n",
      "Time: 2.580472707748413\n",
      "\n",
      "Time: 1.9868428707122803\n",
      "\n",
      "Time: 2.2676098346710205\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.5303797721862793\n",
      "\n",
      "Time: 2.3071255683898926\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.5183262825012207\n",
      "\n",
      "Time: 1.9255688190460205\n",
      "\n",
      "Time: 2.8958358764648438\n",
      "\n",
      "Time: 2.1098105907440186\n",
      "\n",
      "Time: 2.2779085636138916\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.5749142169952393\n",
      "\n",
      "Time: 2.279555082321167\n",
      "\n",
      "Time: 2.629796028137207\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.4965438842773438\n",
      "\n",
      "Time: 2.2775838375091553\n",
      "\n",
      "Time: 2.71659779548645\n",
      "\n",
      "Time: 2.191462993621826\n",
      "\n",
      "Time: 2.628805637359619\n",
      "\n",
      "Time: 1.9191901683807373\n",
      "\n",
      "Time: 2.4842031002044678\n",
      "\n",
      "Time: 2.5166053771972656\n",
      "\n",
      "Time: 2.166447877883911\n",
      "\n",
      "Time: 2.3941640853881836\n",
      "\n",
      "Time: 1.9022841453552246\n",
      "\n",
      "Time: 2.4899544715881348\n",
      "\n",
      "Time: 2.557339668273926\n",
      "\n",
      "Time: 2.2597131729125977\n",
      "\n",
      "Time: 2.534168243408203\n",
      "\n",
      "Time: 2.287539482116699\n",
      "\n",
      "Time: 2.561946392059326\n",
      "\n",
      "Time: 2.086841106414795\n",
      "\n",
      "Time: 2.6206109523773193\n",
      "\n",
      "Time: 2.513747453689575\n",
      "\n",
      "Time: 2.3901748657226562\n",
      "\n",
      "Time: 2.5918385982513428\n",
      "\n",
      "Time: 2.239313840866089\n",
      "\n",
      "Time: 2.809548854827881\n",
      "\n",
      "Time: 2.7469751834869385\n",
      "\n",
      "Time: 2.2950141429901123\n",
      "\n",
      "Time: 2.625988006591797\n",
      "\n",
      "Time: 2.0006887912750244\n",
      "\n",
      "Time: 2.70859956741333\n",
      "\n",
      "Time: 1.9605824947357178\n",
      "\n",
      "Time: 2.3598086833953857\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.4638166427612305\n",
      "\n",
      "Time: 2.3313255310058594\n",
      "\n",
      "Time: 2.5002381801605225\n",
      "\n",
      "Time: 2.317396402359009\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.624452590942383\n",
      "\n",
      "Time: 2.093494176864624\n",
      "\n",
      "Time: 2.3732099533081055\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.519057035446167\n",
      "\n",
      "Time: 2.2385497093200684\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.518535614013672\n",
      "\n",
      "Time: 2.1100881099700928\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.1038897037506104\n",
      "\n",
      "Time: 1.8902966976165771\n",
      "\n",
      "Time: 2.3244259357452393\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.3915998935699463\n",
      "\n",
      "Time: 2.276236057281494\n",
      "\n",
      "Time: 2.4216148853302\n",
      "\n",
      "Time: 1.95577073097229\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.426583766937256\n",
      "\n",
      "Time: 1.8471801280975342\n",
      "\n",
      "Time: 2.4292259216308594\n",
      "\n",
      "Time: 2.3690011501312256\n",
      "\n",
      "Time: 2.0698370933532715\n",
      "\n",
      "Time: 2.3731672763824463\n",
      "\n",
      "Time: 2.232344150543213\n",
      "\n",
      "Time: 2.4207046031951904\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.8191587924957275\n",
      "\n",
      "Time: 2.2325692176818848\n",
      "\n",
      "Time: 2.6194610595703125\n",
      "\n",
      "Time: 2.0561158657073975\n",
      "\n",
      "Time: 2.3727610111236572\n",
      "\n",
      "Time: 2.371978759765625\n",
      "\n",
      "Time: 2.2337653636932373\n",
      "\n",
      "Time: 2.5663230419158936\n",
      "\n",
      "Time: 2.0533816814422607\n",
      "\n",
      "Time: 2.5097880363464355\n",
      "\n",
      "Time: 2.518064498901367\n",
      "\n",
      "Time: 2.319547653198242\n",
      "\n",
      "Time: 2.566708564758301\n",
      "\n",
      "Time: 2.062512159347534\n",
      "\n",
      "Time: 2.6225991249084473\n",
      "\n",
      "Time: 2.5584399700164795\n",
      "\n",
      "Time: 2.1881372928619385\n",
      "\n",
      "Time: 2.4177534580230713\n",
      "\n",
      "Time: 2.248551607131958\n",
      "\n",
      "Time: 2.371579647064209\n",
      "\n",
      "Time: 1.8533215522766113\n",
      "\n",
      "Time: 2.4808788299560547\n",
      "\n",
      "Time: 2.5733091831207275\n",
      "\n",
      "Time: 2.0275349617004395\n",
      "\n",
      "Time: 2.374767780303955\n",
      "\n",
      "Time: 1.898397445678711\n",
      "\n",
      "Time: 2.3799285888671875\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.620708703994751\n",
      "\n",
      "Time: 2.182457685470581\n",
      "\n",
      "Time: 2.4468491077423096\n",
      "\n",
      "Time: 1.9832878112792969\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.61042857170105\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.7551894187927246\n",
      "\n",
      "Time: 2.559438467025757\n",
      "\n",
      "Time: 2.853097915649414\n",
      "\n",
      "Time: 2.237277030944824\n",
      "\n",
      "Time: 2.6790223121643066\n",
      "\n",
      "Time: 2.065248489379883\n",
      "\n",
      "Time: 2.392308473587036\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 3.1624374389648438\n",
      "\n",
      "Time: 2.177893877029419\n",
      "\n",
      "Time: 2.437223434448242\n",
      "\n",
      "Time: 2.125791549682617\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.4024598598480225\n",
      "\n",
      "Time: 1.9754137992858887\n",
      "\n",
      "Time: 2.276840925216675\n",
      "\n",
      "Time: 2.552828550338745\n",
      "\n",
      "Time: 2.161890745162964\n",
      "\n",
      "Time: 2.434418201446533\n",
      "\n",
      "Time: 1.8059372901916504\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.7177393436431885\n",
      "\n",
      "Time: 2.516848087310791\n",
      "\n",
      "Time: 2.1920626163482666\n",
      "\n",
      "Time: 2.4804608821868896\n",
      "\n",
      "Time: 2.206873893737793\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.389843225479126\n",
      "\n",
      "Time: 1.8556489944458008\n",
      "\n",
      "Time: 2.190796136856079\n",
      "\n",
      "Time: 2.459735155105591\n",
      "\n",
      "Time: 2.102717161178589\n",
      "\n",
      "Time: 2.5428974628448486\n",
      "\n",
      "Time: 2.069857358932495\n",
      "\n",
      "Time: 2.551340341567993\n",
      "\n",
      "Time: 2.471088409423828\n",
      "\n",
      "Time: 2.4444797039031982\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.565580368041992\n",
      "\n",
      "Time: 2.1616575717926025\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.927673101425171\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.457263231277466\n",
      "\n",
      "Time: 2.333871841430664\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.2513229846954346\n",
      "\n",
      "Time: 1.9208505153656006\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.380497694015503\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.764342784881592\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Time: 2.415839672088623\n"
     ]
    }
   ],
   "source": [
    "query = raw_data.writeStream\\\n",
    "            .outputMode(\"update\")\\\n",
    "            .foreachBatch(batch_processing)\\\n",
    "            .option(\"checkpointLocation\", \"checkpoint\")\\\n",
    "            .trigger(processingTime='5 seconds')\\\n",
    "            .start()\n",
    "query.awaitTermination(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e255c871",
   "metadata": {},
   "source": [
    "### Metrics\n",
    "\n",
    "To analyze the performance of the various combinations of metrics, we save the metrics in a .json file so they can be analyzed and compared later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "65f31a5f",
   "metadata": {},
   "outputs": [],
   "source": [
    "a = query.recentProgress\n",
    "\n",
    "#kp = kafkapartitions\n",
    "#em exectutor memory\n",
    "#dm driver memory\n",
    "#sp shuffle partitions\n",
    "#aT arrow true\n",
    "#w workers\n",
    "with open('metriche/10ex_1core_6sp_aT_2w_6kp_5000batch_1secProcTime.json', 'w') as file:\n",
    "    # Perform file operations\n",
    "    json.dump(a, file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dd78e075",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "37b68d53",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Print this to stop the code and avoid to stop spark accidentally\n"
     ]
    }
   ],
   "source": [
    "print(\"Print this to stop the code and avoid to stop spark accidentally\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5003bb3d",
   "metadata": {},
   "source": [
    "### Stop workers and master"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "e43c7037",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# stop the running Spark context and Spark session\n",
    "sc.stop()\n",
    "\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eecad1a7",
   "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.8.17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}