StreamProcessing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:17:47.067778Z",
     "start_time": "2021-06-01T09:17:46.964791Z"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as f\n",
    "from pyspark.sql import types as t\n",
    "from IPython.display import display, clear_output\n",
    "from datetime import datetime\n",
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:17:47.117299Z",
     "start_time": "2021-06-01T09:17:47.113996Z"
    }
   },
   "outputs": [],
   "source": [
    "from os import environ\n",
    "environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 pyspark-shell'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:17:52.272282Z",
     "start_time": "2021-06-01T09:17:47.261194Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.1.1'"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession.builder.appName(\"StructuredStreaming\").getOrCreate()\n",
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Import rozkładu przystanków z pliku"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-24T19:47:05.417798Z",
     "start_time": "2021-05-24T19:47:04.910453Z"
    }
   },
   "outputs": [],
   "source": [
    "# df_bus = spark.read.format('parquet').load('df_bus.parquet')\n",
    "# df_bus.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:17:56.791521Z",
     "start_time": "2021-06-01T09:17:52.365458Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "209517"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_tram_tmp = spark.read.format('parquet').load('./Schedule/df_tram.parquet')\n",
    "df_tram_tmp.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-24T19:47:16.280728Z",
     "start_time": "2021-05-24T19:47:16.009407Z"
    }
   },
   "outputs": [],
   "source": [
    "# df_all = spark.read.format('parquet').load('df_all.parquet')\n",
    "# df_all.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:21:12.460735Z",
     "start_time": "2021-06-01T09:21:12.455363Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- line: long (nullable = true)\n",
      " |-- busStopName: string (nullable = true)\n",
      " |-- sLat: double (nullable = true)\n",
      " |-- sLon: double (nullable = true)\n",
      " |-- sTime: string (nullable = true)\n",
      " |-- vehicleID: string (nullable = true)\n",
      " |-- busStopID: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_tram_tmp.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:18:57.126414Z",
     "start_time": "2021-06-01T09:18:56.633893Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------+---------+---------+--------+---------+---------+\n",
      "|line|busStopName|     sLat|     sLon|   sTime|vehicleID|busStopID|\n",
      "+----+-----------+---------+---------+--------+---------+---------+\n",
      "|  25|Poborzańska|52.289921|21.029526|04:54:00|     25/4|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|05:14:00|     25/5|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|05:24:00|     25/8|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|05:35:00|   25/014|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|05:54:00|    25/11|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|06:04:00|    25/12|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|06:14:00|     25/2|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|06:34:00|   25/013|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|06:42:00|     25/3|  1084/02|\n",
      "|  25|Poborzańska|52.289921|21.029526|06:55:00|     25/4|  1084/02|\n",
      "+----+-----------+---------+---------+--------+---------+---------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_tram_tmp.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dodatkowe obrobienie df_tram"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:19:52.640974Z",
     "start_time": "2021-06-01T09:19:52.630503Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "datetime.datetime(2021, 6, 1, 14, 54)"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test = '14:54:00'\n",
    "def convDateTime(x):\n",
    "    strDT =  f\"{datetime.strftime(datetime.date(datetime.now()),'%Y-%m-%d')} {x}\" \n",
    "    \n",
    "    tmpMonth = int(strDT[5:7])\n",
    "    tmpDay = int(strDT[8:10])\n",
    "    tmpHour = int(strDT[11:13])\n",
    "        \n",
    "    if tmpHour != 24:\n",
    "        return datetime.strptime(strDT,'%Y-%m-%d %H:%M:%S')\n",
    "    else:\n",
    "        newHour = '00'\n",
    "        newDay = str(tmpDay + 1)\n",
    "        if newDay != '32':\n",
    "            resStr = strDT[:8] + newDay + strDT[10:11] + newHour + strDT[13:]\n",
    "        else:\n",
    "            newDay = '01'\n",
    "            newMonth = str(tmpMonth + 1)\n",
    "            resStr = strDT[:5] + newMonth + strDT[7:8] + newDay + strDT[10:11] + newHour + strDT[13:]\n",
    "            \n",
    "        return datetime.strptime(resStr,'%Y-%m-%d %H:%M:%S')\n",
    "    \n",
    "    \n",
    "convDateTime(test)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:19:55.305683Z",
     "start_time": "2021-06-01T09:19:55.302869Z"
    }
   },
   "outputs": [],
   "source": [
    "def secsFrom1970(x):\n",
    "    epoch = datetime.utcfromtimestamp(0)\n",
    "    return (x - epoch).total_seconds()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:19:56.065748Z",
     "start_time": "2021-06-01T09:19:56.062892Z"
    }
   },
   "outputs": [],
   "source": [
    "udfSecsFrom1970 = f.udf(secsFrom1970, t.FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-08T19:44:29.288668Z",
     "start_time": "2021-06-08T19:44:29.283440Z"
    }
   },
   "outputs": [],
   "source": [
    "import math\n",
    "def measureDist(lat1, lon1, lat2, lon2) :\n",
    "    R = 6378.137  # Radius of earth in KM\n",
    "    dLat = lat2 * math.pi / 180 - lat1 * math.pi / 180\n",
    "    dLon = lon2 * math.pi / 180 - lon1 * math.pi / 180\n",
    "    a = math.sin(dLat/2) * math.sin(dLat/2) + \\\n",
    "        math.cos(lat1 * math.pi / 180) * math.cos(lat2 * math.pi / 180) * \\\n",
    "        math.sin(dLon/2) * math.sin(dLon/2)\n",
    "    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))\n",
    "    d = R * c\n",
    "    return d * 1000  # meters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-08T19:49:51.870994Z",
     "start_time": "2021-06-08T19:49:51.865295Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5.565974539011866"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "measureDist(52.28995, 21.02957, 52.28990, 21.02957)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:20:04.177164Z",
     "start_time": "2021-06-01T09:20:04.174097Z"
    }
   },
   "outputs": [],
   "source": [
    "udfMeasureDist = f.udf(measureDist, t.FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:22:09.578610Z",
     "start_time": "2021-06-01T09:22:09.574092Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- line: long (nullable = true)\n",
      " |-- busStopName: string (nullable = true)\n",
      " |-- sLat: double (nullable = true)\n",
      " |-- sLon: double (nullable = true)\n",
      " |-- sTime: string (nullable = true)\n",
      " |-- vehicleID: string (nullable = true)\n",
      " |-- busStopID: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_tram_tmp.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:23:29.035278Z",
     "start_time": "2021-06-01T09:23:28.902442Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- line: long (nullable = true)\n",
      " |-- busStopName: string (nullable = true)\n",
      " |-- sLat: double (nullable = true)\n",
      " |-- sLon: double (nullable = true)\n",
      " |-- sTime: timestamp (nullable = true)\n",
      " |-- sVehicleID: string (nullable = true)\n",
      " |-- busStopID: string (nullable = true)\n",
      " |-- sTimer: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rdd1 = df_tram_tmp.rdd.map(list)\n",
    "rdd2 = rdd1.map(lambda x: [x[0], x[1], x[2], x[3], convDateTime(x[4]), x[5], x[6]])\n",
    "rdd3 = rdd2.map(lambda x: [x[0], x[1], x[2], x[3], x[4], x[5], x[6], secsFrom1970(x[4])])\n",
    "# rdd2 = rdd1.map(lambda x: [x[0], x[1], x[2], f\"{datetime.strftime(datetime.date(datetime.now()),'%Y-%m-%d')} {x[3]}\", x[4], x[5]])\n",
    "# rdd3 = rdd2.map(lambda x: [x[0], x[1], x[2], datetime.strptime(x[3],'%Y-%m-%d %H:%M:%S'), x[4], x[5]])\n",
    "# rdd4 = rdd3.map(lambda x: [x[0], x[1], x[2], x[3], x[4], x[5], datetime.utcfromtimestamp(x[3])])\n",
    "# df_tram = rdd4.toDF(['busStopName', 'sLat', 'sLon', 'sTime', 'vehicleID', 'busStopID', 'sEpoch'])\n",
    "df_tram = rdd3.toDF(['line','busStopName', 'sLat', 'sLon', 'sTime', 'sVehicleID', 'busStopID', 'sTimer'])\n",
    "df_tram.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:23:30.461539Z",
     "start_time": "2021-06-01T09:23:30.254090Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----------+---------+---------+-------------------+----------+---------+------------+\n",
      "|line|busStopName|     sLat|     sLon|              sTime|sVehicleID|busStopID|      sTimer|\n",
      "+----+-----------+---------+---------+-------------------+----------+---------+------------+\n",
      "|  25|Poborzańska|52.289921|21.029526|2021-06-01 04:54:00|      25/4|  1084/02|1.62252324E9|\n",
      "|  25|Poborzańska|52.289921|21.029526|2021-06-01 05:14:00|      25/5|  1084/02|1.62252444E9|\n",
      "|  25|Poborzańska|52.289921|21.029526|2021-06-01 05:24:00|      25/8|  1084/02|1.62252504E9|\n",
      "|  25|Poborzańska|52.289921|21.029526|2021-06-01 05:35:00|    25/014|  1084/02| 1.6225257E9|\n",
      "|  25|Poborzańska|52.289921|21.029526|2021-06-01 05:54:00|     25/11|  1084/02|1.62252684E9|\n",
      "+----+-----------+---------+---------+-------------------+----------+---------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_tram.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:23:40.403017Z",
     "start_time": "2021-06-01T09:23:38.061626Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "209517"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_tram.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Test kafki"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-21T18:30:59.395880Z",
     "start_time": "2021-05-21T18:30:59.392769Z"
    },
    "heading_collapsed": true
   },
   "source": [
    "## Uruchomienie pracy równoległej"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-24T20:00:44.337428Z",
     "start_time": "2021-05-24T20:00:34.304971Z"
    },
    "hidden": true,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import ipyparallel as ipp\n",
    "\n",
    "rc = ipp.Client(profile='default')\n",
    "ar = rc[:].apply_async(os.getpid)\n",
    "pid_map = ar.get_dict()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-24T20:00:44.339904Z",
     "start_time": "2021-05-24T20:00:37.477Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "rc.ids"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Subskrybcja topicu tram z Kafki"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:20.283671Z",
     "start_time": "2021-06-01T09:57:20.280319Z"
    }
   },
   "outputs": [],
   "source": [
    "# bootstrap_servers = 'kafka-1:19091,kafka-2:29091,kafka-3:39091'\n",
    "# bootstrap_servers = 'localhost:19092,localhost:29092,localhost:39092'\n",
    "bootstrap_servers = 'localhost:19092'\n",
    "topic = 'tram'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:20.690307Z",
     "start_time": "2021-06-01T09:57:20.654777Z"
    }
   },
   "outputs": [],
   "source": [
    "raw_df = spark \\\n",
    "    .readStream \\\n",
    "    .format(\"kafka\") \\\n",
    "    .option(\"kafka.bootstrap.servers\", bootstrap_servers) \\\n",
    "    .option(\"subscribe\", topic) \\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:23.226688Z",
     "start_time": "2021-06-01T09:57:23.221800Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- key: binary (nullable = true)\n",
      " |-- value: binary (nullable = true)\n",
      " |-- topic: string (nullable = true)\n",
      " |-- partition: integer (nullable = true)\n",
      " |-- offset: long (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      " |-- timestampType: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "raw_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:26.450067Z",
     "start_time": "2021-06-01T09:57:26.446492Z"
    }
   },
   "outputs": [],
   "source": [
    "# {\n",
    "#     \"Lines\": \"213\",\n",
    "#     \"Lon\": 21.1291008,\n",
    "#     \"VehicleNumber\": \"1000\",\n",
    "#     \"Time\": \"2021-05-21 20:17:34\",\n",
    "#     \"Lat\": 52.2146371,\n",
    "#     \"Brigade\": \"2\"\n",
    "# }\n",
    "\n",
    "schema = t.StructType() \\\n",
    "    .add(\"Lines\", t.StringType()) \\\n",
    "    .add(\"Lon\", t.FloatType()) \\\n",
    "    .add(\"VehicleNumber\", t.StringType()) \\\n",
    "    .add(\"Time\", t.TimestampType()) \\\n",
    "    .add(\"Lat\", t.FloatType()) \\\n",
    "    .add(\"Brigade\", t.StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:27.179794Z",
     "start_time": "2021-06-01T09:57:26.839629Z"
    }
   },
   "outputs": [],
   "source": [
    "stream_tram = raw_df.select( \\\n",
    "    raw_df.key.cast('string'),\n",
    "    f.from_json(raw_df.value.cast('string'), schema))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:27.869212Z",
     "start_time": "2021-06-01T09:57:27.854208Z"
    }
   },
   "outputs": [],
   "source": [
    "stream_tram = stream_tram.withColumnRenamed('from_json(CAST(value AS STRING))', 'json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:29.756645Z",
     "start_time": "2021-06-01T09:57:29.581250Z"
    }
   },
   "outputs": [],
   "source": [
    "stream_tram = stream_tram \\\n",
    "    .withColumnRenamed('key', 'vVehicleID') \\\n",
    "    .withColumn('vLon', stream_tram.json.Lon) \\\n",
    "    .withColumn('vLat', stream_tram.json.Lat) \\\n",
    "    .withColumn('vTime', stream_tram.json.Time)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:30.482552Z",
     "start_time": "2021-06-01T09:57:30.469756Z"
    }
   },
   "outputs": [],
   "source": [
    "stream_tram = stream_tram.drop(stream_tram.json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:31.126856Z",
     "start_time": "2021-06-01T09:57:31.092564Z"
    }
   },
   "outputs": [],
   "source": [
    "stream_tram = stream_tram.withColumn('vTimer', udfSecsFrom1970(f.col('vTime')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:32.665613Z",
     "start_time": "2021-06-01T09:57:32.661305Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- line: long (nullable = true)\n",
      " |-- busStopName: string (nullable = true)\n",
      " |-- sLat: double (nullable = true)\n",
      " |-- sLon: double (nullable = true)\n",
      " |-- sTime: timestamp (nullable = true)\n",
      " |-- sVehicleID: string (nullable = true)\n",
      " |-- busStopID: string (nullable = true)\n",
      " |-- sTimer: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_tram.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T09:57:33.578399Z",
     "start_time": "2021-06-01T09:57:33.574230Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- vVehicleID: string (nullable = true)\n",
      " |-- vLon: float (nullable = true)\n",
      " |-- vLat: float (nullable = true)\n",
      " |-- vTime: timestamp (nullable = true)\n",
      " |-- vTimer: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "stream_tram.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Utworzenie query do sprawdzania opóźnień na przystankach"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:09:16.137733Z",
     "start_time": "2021-06-01T10:09:16.113356Z"
    }
   },
   "outputs": [],
   "source": [
    "accDiff = 0.00005\n",
    "cond1 = [df_tram.sVehicleID == stream_tram.vVehicleID, \\\n",
    "        df_tram.sLat - stream_tram.vLat < accDiff, \\\n",
    "        df_tram.sLat - stream_tram.vLat > -accDiff, \\\n",
    "        df_tram.sLon - stream_tram.vLon < accDiff, \\\n",
    "        df_tram.sLon - stream_tram.vLon > -accDiff ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:10:45.022449Z",
     "start_time": "2021-06-01T10:10:44.975023Z"
    }
   },
   "outputs": [],
   "source": [
    "query1 = stream_tram.join(df_tram, cond1)\n",
    "query1 = query1.withColumn('del', f.abs(query1.sTimer - query1.vTimer)).groupBy(f.col('vVehicleID'), f.col('busStopName'), f.col('busStopID'), f.col('line')).agg(f.min('del').alias('delay'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Utworzenie query do sprawdzenia odległości pojazdu od przystanku"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-26T14:41:56.456172Z",
     "start_time": "2021-05-26T14:41:56.449109Z"
    }
   },
   "outputs": [],
   "source": [
    "cond2 = [df_tram.sVehicleID == stream_tram.vVehicleID, \\\n",
    "        df_tram.sTime == stream_tram.vTime ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-26T14:41:57.335510Z",
     "start_time": "2021-05-26T14:41:57.257213Z"
    }
   },
   "outputs": [],
   "source": [
    "# measureDist(lat1, lon1, lat2, lon2)\n",
    "query2 = stream_tram.join(df_tram, cond2).withColumn('distanceInMeters', udfMeasureDist(f.col('sLat'), f.col('sLon'), f.col('vLat'), f.col('vLon')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:10:48.747498Z",
     "start_time": "2021-06-01T10:10:48.742565Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "True"
      ]
     },
     "execution_count": 57,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "datetime(2021, 5, 26, 4, 54) == datetime(2021, 5, 26, 4, 54)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:00:08.210226Z",
     "start_time": "2021-06-01T09:59:37.876991Z"
    }
   },
   "outputs": [],
   "source": [
    "q = stream_tram \\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"memory\") \\\n",
    "    .queryName(\"myQuery\") \\\n",
    "    .start()\n",
    "\n",
    "q.awaitTermination(30)\n",
    "q.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:11:52.122532Z",
     "start_time": "2021-06-01T10:10:52.070189Z"
    }
   },
   "outputs": [],
   "source": [
    "q1 = query1 \\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"update\") \\\n",
    "    .format(\"memory\") \\\n",
    "    .queryName(\"myQuery\") \\\n",
    "    .start()\n",
    "\n",
    "q1.awaitTermination(60)\n",
    "q1.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-26T14:45:13.596283Z",
     "start_time": "2021-05-26T14:44:59.828651Z"
    },
    "collapsed": true
   },
   "outputs": [
    {
     "ename": "StreamingQueryException",
     "evalue": "Writing job aborted.\n=== Streaming Query ===\nIdentifier: myQuery [id = f2ace432-60a5-4af2-9c7d-dcd4ddb550e9, runId = fc04ec78-b6b3-47d4-8a45-54156d9af735]\nCurrent Committed Offsets: {KafkaV2[Subscribe[tram]]: {\"tram\":{\"0\":2277103}}}\nCurrent Available Offsets: {KafkaV2[Subscribe[tram]]: {\"tram\":{\"0\":2277400}}}\n\nCurrent State: ACTIVE\nThread State: RUNNABLE\n\nLogical Plan:\nWriteToMicroBatchDataSource org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@41bd8fc8\n+- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204, vTimer#215, busStopName#109, sLat#110, sLon#111, sTime#112, sVehicleID#113, busStopID#114, sTimer#115, measureDist(sLat#110, sLon#111, vLat#199, vLon#195) AS distanceInMeters#26450]\n   +- Join Inner, ((sVehicleID#113 = vVehicleID#192) AND (sTime#112 = vTime#204))\n      :- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204, secsFrom1970(vTime#204) AS vTimer#215]\n      :  +- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204]\n      :     +- Project [vVehicleID#192, json#189, vLon#195, vLat#199, json#189.Time AS vTime#204]\n      :        +- Project [vVehicleID#192, json#189, vLon#195, json#189.Lat AS vLat#199]\n      :           +- Project [vVehicleID#192, json#189, json#189.Lon AS vLon#195]\n      :              +- Project [key#186 AS vVehicleID#192, json#189]\n      :                 +- Project [key#186, from_json(CAST(value AS STRING))#185 AS json#189]\n      :                    +- Project [cast(key#171 as string) AS key#186, from_json(StructField(Lines,StringType,true), StructField(Lon,FloatType,true), StructField(VehicleNumber,StringType,true), StructField(Time,TimestampType,true), StructField(Lat,FloatType,true), StructField(Brigade,StringType,true), cast(value#172 as string), Some(Europe/Warsaw)) AS from_json(CAST(value AS STRING))#185]\n      :                       +- StreamingDataSourceV2Relation [key#171, value#172, topic#173, partition#174, offset#175L, timestamp#176, timestampType#177], org.apache.spark.sql.kafka010.KafkaSourceProvider$KafkaScan@aa25b06, KafkaV2[Subscribe[tram]]\n      +- LogicalRDD [busStopName#109, sLat#110, sLon#111, sTime#112, sVehicleID#113, busStopID#114, sTimer#115], false\n",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mStreamingQueryException\u001b[0m                   Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-45-5a1bf06c9891>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      6\u001b[0m     \u001b[0;34m.\u001b[0m\u001b[0mstart\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[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 8\u001b[0;31m \u001b[0mq2\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mawaitTermination\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m120\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      9\u001b[0m \u001b[0mq2\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstop\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[0;32m~/anaconda3/envs/spark/lib/python3.9/site-packages/pyspark/sql/streaming.py\u001b[0m in \u001b[0;36mawaitTermination\u001b[0;34m(self, timeout)\u001b[0m\n\u001b[1;32m     97\u001b[0m             \u001b[0;32mif\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtimeout\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m(\u001b[0m\u001b[0mint\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mfloat\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0mtimeout\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[1;32m     98\u001b[0m                 \u001b[0;32mraise\u001b[0m \u001b[0mValueError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"timeout must be a positive integer or float. Got %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mtimeout\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 99\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jsq\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mawaitTermination\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mtimeout\u001b[0m \u001b[0;34m*\u001b[0m \u001b[0;36m1000\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[0m\u001b[1;32m    100\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    101\u001b[0m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jsq\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mawaitTermination\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[0;32m~/anaconda3/envs/spark/lib/python3.9/site-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1302\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1303\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[0;32m-> 1304\u001b[0;31m         return_value = get_return_value(\n\u001b[0m\u001b[1;32m   1305\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[1;32m   1306\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/anaconda3/envs/spark/lib/python3.9/site-packages/pyspark/sql/utils.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    115\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    116\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--> 117\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    118\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    119\u001b[0m                 \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mStreamingQueryException\u001b[0m: Writing job aborted.\n=== Streaming Query ===\nIdentifier: myQuery [id = f2ace432-60a5-4af2-9c7d-dcd4ddb550e9, runId = fc04ec78-b6b3-47d4-8a45-54156d9af735]\nCurrent Committed Offsets: {KafkaV2[Subscribe[tram]]: {\"tram\":{\"0\":2277103}}}\nCurrent Available Offsets: {KafkaV2[Subscribe[tram]]: {\"tram\":{\"0\":2277400}}}\n\nCurrent State: ACTIVE\nThread State: RUNNABLE\n\nLogical Plan:\nWriteToMicroBatchDataSource org.apache.spark.sql.execution.streaming.sources.MemoryStreamingWrite@41bd8fc8\n+- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204, vTimer#215, busStopName#109, sLat#110, sLon#111, sTime#112, sVehicleID#113, busStopID#114, sTimer#115, measureDist(sLat#110, sLon#111, vLat#199, vLon#195) AS distanceInMeters#26450]\n   +- Join Inner, ((sVehicleID#113 = vVehicleID#192) AND (sTime#112 = vTime#204))\n      :- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204, secsFrom1970(vTime#204) AS vTimer#215]\n      :  +- Project [vVehicleID#192, vLon#195, vLat#199, vTime#204]\n      :     +- Project [vVehicleID#192, json#189, vLon#195, vLat#199, json#189.Time AS vTime#204]\n      :        +- Project [vVehicleID#192, json#189, vLon#195, json#189.Lat AS vLat#199]\n      :           +- Project [vVehicleID#192, json#189, json#189.Lon AS vLon#195]\n      :              +- Project [key#186 AS vVehicleID#192, json#189]\n      :                 +- Project [key#186, from_json(CAST(value AS STRING))#185 AS json#189]\n      :                    +- Project [cast(key#171 as string) AS key#186, from_json(StructField(Lines,StringType,true), StructField(Lon,FloatType,true), StructField(VehicleNumber,StringType,true), StructField(Time,TimestampType,true), StructField(Lat,FloatType,true), StructField(Brigade,StringType,true), cast(value#172 as string), Some(Europe/Warsaw)) AS from_json(CAST(value AS STRING))#185]\n      :                       +- StreamingDataSourceV2Relation [key#171, value#172, topic#173, partition#174, offset#175L, timestamp#176, timestampType#177], org.apache.spark.sql.kafka010.KafkaSourceProvider$KafkaScan@aa25b06, KafkaV2[Subscribe[tram]]\n      +- LogicalRDD [busStopName#109, sLat#110, sLon#111, sTime#112, sVehicleID#113, busStopID#114, sTimer#115], false\n"
     ]
    }
   ],
   "source": [
    "q2 = query2 \\\n",
    "    .writeStream \\\n",
    "    .outputMode(\"append\") \\\n",
    "    .format(\"memory\") \\\n",
    "    .queryName(\"myQuery\") \\\n",
    "    .start()\n",
    "\n",
    "q2.awaitTermination(120)\n",
    "q2.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-06-01T10:11:52.153799Z",
     "start_time": "2021-06-01T10:11:52.125592Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+---------+----+-----+\n",
      "|vVehicleID|busStopName|busStopID|line|delay|\n",
      "+----------+-----------+---------+----+-----+\n",
      "+----------+-----------+---------+----+-----+\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "None"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(spark.sql('SELECT * FROM myQuery').show())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "heading_collapsed": true
   },
   "source": [
    "## Chyba zbędne"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-21T18:56:55.961834Z",
     "start_time": "2021-05-21T18:56:39.449413Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "while True:\n",
    "    clear_output(wait=True)\n",
    "    display(q1.status)\n",
    "    display(spark.sql('SELECT * FROM myQuery').show())\n",
    "    sleep(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T22:25:34.359892Z",
     "start_time": "2021-05-20T22:25:34.356483Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "__.display_outputs()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T14:47:21.660478Z",
     "start_time": "2021-05-20T14:47:21.572046Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "raw.selectExpr(\"CAST(key AS STRING)\", \"CAST(value AS STRING)\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T14:42:06.185357Z",
     "start_time": "2021-05-20T14:42:06.118301Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "q1 = raw.writeStream.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T14:23:35.235641Z",
     "start_time": "2021-05-20T14:23:32.078429Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "query = raw.writeStream.outputMode(\"append\").format(\"console\").start()\n",
    "query.awaitTermination(60)\n",
    "query.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T11:49:02.076278Z",
     "start_time": "2021-05-20T11:49:01.812072Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "words = raw.select(f.explode(f.split(raw.value, \" \")).alias(\"word\"))\n",
    "wordCounts = words.groupBy(\"word\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-05-20T11:49:06.798287Z",
     "start_time": "2021-05-20T11:49:03.179422Z"
    },
    "hidden": true
   },
   "outputs": [],
   "source": [
    "query = wordCounts.writeStream.outputMode(\"complete\").format(\"console\").start()\n",
    "query.awaitTermination(60)\n",
    "query.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.4 (spark)",
   "language": "python",
   "name": "spark"
  },
  "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.9.4"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}