spark_stream_twitter.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "28b5a46a",
   "metadata": {},
   "outputs": [
    {
     "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"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:13:54 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/context.py:114: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.\n",
      "  FutureWarning,\n"
     ]
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.functions import desc\n",
    "\n",
    "sc = SparkContext()\n",
    "\n",
    "ssc = StreamingContext(sc, 10)\n",
    "sqlcontext = SQLContext(sc)\n",
    "socket_stream = ssc.socketTextStream('127.0.0.1', 5554)\n",
    "lines = socket_stream.window(20)\n",
    "from collections import namedtuple\n",
    "fields = ('tags',\"count\")\n",
    "Tweet = namedtuple ('Tweet', fields)\n",
    "( lines.flatMap( lambda text: text.split( \" \" ) )                 # Splits to a list\n",
    "  .filter( lambda word: word.lower().startswith(\"#\") )            # Checks for hashtag calls\n",
    "  .map( lambda word: ( word.lower(), 1 ) )                        # Lower cases the word\n",
    "  .reduceByKey( lambda a, b: a + b )                              # Reduces\n",
    "  .map( lambda rec: Tweet( rec[0], rec[1] ) )                     # Stores in a Tweet Object\n",
    "  .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
    "  .limit(10).registerTempTable(\"tweets\") ) )                      # Registers to a table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "5eef2a35",
   "metadata": {},
   "outputs": [],
   "source": [
    "ssc.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "0ff40ca3",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 0:>                                                          (0 + 1) / 1]\r"
     ]
    }
   ],
   "source": [
    "import time\n",
    "from IPython import display\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "# Only works for Jupyter Notebooks!\n",
    "%matplotlib inline "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2187e1bc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:03 ERROR JobScheduler: Error running job streaming job 1664086440000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 0:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "ename": "AnalysisException",
     "evalue": "Table or view not found: tweets; line 1 pos 23;\n'Project ['tag, 'count]\n+- 'UnresolvedRelation [tweets], [], false\n",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "\u001b[0;32m/tmp/ipykernel_7771/3233657615.py\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      3\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m     \u001b[0mtime\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[0;34m(\u001b[0m \u001b[0;36m3\u001b[0m \u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 5\u001b[0;31m     \u001b[0mtop_10_tweets\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0msqlcontext\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m\u001b[0;34m(\u001b[0m \u001b[0;34m'Select tag, count from tweets'\u001b[0m \u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      6\u001b[0m     \u001b[0mtop_10_df\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mtop_10_tweets\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoPandas\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      7\u001b[0m     \u001b[0mdisplay\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mclear_output\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mwait\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/context.py\u001b[0m in \u001b[0;36msql\u001b[0;34m(self, sqlQuery)\u001b[0m\n\u001b[1;32m    543\u001b[0m         \u001b[0;34m[\u001b[0m\u001b[0mRow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m1\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'row1'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mRow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m2\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'row2'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mRow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mf1\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;36m3\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mf2\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m'row3'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    544\u001b[0m         \"\"\"\n\u001b[0;32m--> 545\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msparkSession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msqlQuery\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    546\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    547\u001b[0m     \u001b[0;32mdef\u001b[0m \u001b[0mtable\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtableName\u001b[0m\u001b[0;34m:\u001b[0m \u001b[0mstr\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m->\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\u001b[0m in \u001b[0;36msql\u001b[0;34m(self, sqlQuery, **kwargs)\u001b[0m\n\u001b[1;32m   1032\u001b[0m             \u001b[0msqlQuery\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mformatter\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msqlQuery\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m**\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1033\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1034\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mDataFrame\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jsparkSession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msql\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msqlQuery\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1035\u001b[0m         \u001b[0;32mfinally\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1036\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0mlen\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mkwargs\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m>\u001b[0m \u001b[0;36m0\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1320\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1321\u001b[0m         return_value = get_return_value(\n\u001b[0;32m-> 1322\u001b[0;31m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[0m\u001b[1;32m   1323\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1324\u001b[0m         \u001b[0;32mfor\u001b[0m \u001b[0mtemp_arg\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mtemp_args\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    194\u001b[0m                 \u001b[0;31m# Hide where the exception came from that shows a non-Pythonic\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    195\u001b[0m                 \u001b[0;31m# JVM exception message.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 196\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mconverted\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    197\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    198\u001b[0m                 \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mAnalysisException\u001b[0m: Table or view not found: tweets; line 1 pos 23;\n'Project ['tag, 'count]\n+- 'UnresolvedRelation [tweets], [], false\n"
     ]
    }
   ],
   "source": [
    "count = 0\n",
    "while count < 10:\n",
    "    \n",
    "    time.sleep( 3 )\n",
    "    top_10_tweets = sqlcontext.sql( 'Select tag, count from tweets' )\n",
    "    top_10_df = top_10_tweets.toPandas()\n",
    "    display.clear_output(wait=True)\n",
    "    plt.figure( figsize = ( 10, 8 ) )\n",
    "#     sns.barplot(x='count',y='land_cover_specific', data=df, palette='Spectral')\n",
    "    sns.barplot( x=\"count\", y=\"tag\", data=top_10_df)\n",
    "    plt.show()\n",
    "    count = count + 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b228f783",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:10 ERROR JobScheduler: Error running job streaming job 1664086450000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:20 ERROR JobScheduler: Error running job streaming job 1664086460000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:30 ERROR JobScheduler: Error running job streaming job 1664086470000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:40 ERROR JobScheduler: Error running job streaming job 1664086480000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:14:50 ERROR JobScheduler: Error running job streaming job 1664086490000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:00 ERROR JobScheduler: Error running job streaming job 1664086500000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:10 ERROR JobScheduler: Error running job streaming job 1664086510000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:20 ERROR JobScheduler: Error running job streaming job 1664086520000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:30 ERROR JobScheduler: Error running job streaming job 1664086530000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:40 ERROR JobScheduler: Error running job streaming job 1664086540000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:15:50 ERROR JobScheduler: Error running job streaming job 1664086550000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:00 ERROR JobScheduler: Error running job streaming job 1664086560000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:10 ERROR JobScheduler: Error running job streaming job 1664086570000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:20 ERROR JobScheduler: Error running job streaming job 1664086580000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:30 ERROR JobScheduler: Error running job streaming job 1664086590000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:40 ERROR JobScheduler: Error running job streaming job 1664086600000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:16:50 ERROR JobScheduler: Error running job streaming job 1664086610000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:00 ERROR JobScheduler: Error running job streaming job 1664086620000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:10 ERROR JobScheduler: Error running job streaming job 1664086630000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:20 ERROR JobScheduler: Error running job streaming job 1664086640000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:30 ERROR JobScheduler: Error running job streaming job 1664086650000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:40 ERROR JobScheduler: Error running job streaming job 1664086660000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:17:50 ERROR JobScheduler: Error running job streaming job 1664086670000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:00 ERROR JobScheduler: Error running job streaming job 1664086680000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:10 ERROR JobScheduler: Error running job streaming job 1664086690000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:20 ERROR JobScheduler: Error running job streaming job 1664086700000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:30 ERROR JobScheduler: Error running job streaming job 1664086710000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:40 ERROR JobScheduler: Error running job streaming job 1664086720000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:18:50 ERROR JobScheduler: Error running job streaming job 1664086730000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/09/25 06:19:00 ERROR JobScheduler: Error running job streaming job 1664086740000 ms.0\n",
      "org.apache.spark.SparkException: An exception was raised by Python:\n",
      "Traceback (most recent call last):\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/util.py\", line 71, in call\n",
      "    r = self.func(t, *rdds)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/streaming/dstream.py\", line 236, in func\n",
      "    return old_func(rdd)  # type: ignore[call-arg, arg-type]\n",
      "  File \"/tmp/ipykernel_7771/455637378.py\", line 20, in <lambda>\n",
      "    .foreachRDD( lambda rdd: rdd.toDF().sort( desc(\"count\") )       # Sorts Them in a DF\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 102, in toDF\n",
      "    return sparkSession.createDataFrame(self, schema, sampleRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 895, in createDataFrame\n",
      "    data, schema, samplingRatio, verifySchema  # type: ignore[arg-type]\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 934, in _create_dataframe\n",
      "    rdd, struct = self._createFromRDD(data.map(prepare), schema, samplingRatio)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 600, in _createFromRDD\n",
      "    struct = self._inferSchema(rdd, samplingRatio, names=schema)\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/sql/session.py\", line 546, in _inferSchema\n",
      "    first = rdd.first()\n",
      "  File \"/home/ec2-user/anaconda3/envs/amazonei_pytorch_latest_p37/lib/python3.7/site-packages/pyspark/rdd.py\", line 1906, in first\n",
      "    raise ValueError(\"RDD is empty\")\n",
      "ValueError: RDD is empty\n",
      "\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.callPythonTransformFunction(PythonDStream.scala:95)\n",
      "\tat org.apache.spark.streaming.api.python.TransformFunction.apply(PythonDStream.scala:78)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.api.python.PythonDStream$.$anonfun$callForeachRDD$1$adapted(PythonDStream.scala:179)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$2(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:417)\n",
      "\tat org.apache.spark.streaming.dstream.ForEachDStream.$anonfun$generateJob$1(ForEachDStream.scala:51)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.Try$.apply(Try.scala:213)\n",
      "\tat org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.$anonfun$run$1(JobScheduler.scala:256)\n",
      "\tat scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)\n",
      "\tat scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)\n",
      "\tat org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)\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:834)\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 0:>                                                          (0 + 1) / 1]\r"
     ]
    }
   ],
   "source": [
    "ssc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5033f62b",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10.5 64-bit",
   "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.10.5"
  },
  "vscode": {
   "interpreter": {
    "hash": "b64e5434eb6f78b23ded7093c27034844fda971e57a0689a97b233b8e8d5583b"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}