DSLab_homework4-3.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DSLab Homework4 - More trains (PART III)\n",
    "\n",
    "## Hand-in Instructions:\n",
    "- __Due: 11.05.2021 23:59:59 CET__\n",
    "- your project must be private\n",
    "- git push your final verion to the master branch of your group's Renku repository before the due date\n",
    "- check if Dockerfile, environment.yml and requirements.txt are properly written\n",
    "- add necessary comments and discussion to make your codes readable\n",
    "\n",
    "## NS Streams\n",
    "For this homework, you will be working with the real-time streams of the NS, the train company of the Netherlands. You can see an example webpage that uses the same streams to display the train information on a map: https://spoorkaart.mwnn.nl/ . \n",
    "\n",
    "To help you and avoid having too many connections to the NS streaming servers, we have setup a service that collects the streams and pushes them to our Kafka instance. The related topics are: \n",
    "\n",
    "`ndovloketnl-arrivals`: For each arrival of a train in a station, describe the previous and next station, time of arrival (planned and actual), track number,...\n",
    "\n",
    "`ndovloketnl-departures`: For each departure of a train from a station, describe the previous and next station, time of departure (planned and actual), track number,...\n",
    "\n",
    "`ndovloketnl-gps`: For each train, describe the current location, speed, bearing.\n",
    "\n",
    "The events are serialized in JSON (actually converted from XML), with properties in their original language. Google translate could help you understand all of them, but we will provide you with some useful mappings."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "**PART III is in PySpark kernel**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Current kernel: PySpark\n"
     ]
    }
   ],
   "source": [
    "%%local\n",
    "ipython = get_ipython()\n",
    "print('Current kernel: {}'.format(ipython.kernel.kernel_info['implementation']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## Set up environment\n",
    "\n",
    "Run the following cells below before running the other cells of this notebook. Run them whenever you need to recreate a Spark context. Pay particular attention to your `username` settings, and make sure that it is properly set to your user name, both locally and on the remote Spark Driver.\n",
    "\n",
    "Configure your spark settings:\n",
    "1. name your spark application as `\"<your_gaspar_id>-homework4\"`.\n",
    "2. make the required kafka jars available on the remote Spark driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{'name': 'cadei-homework4', 'executorMemory': '1G', 'executorCores': 2, 'numExecutors': 2, 'conf': {'spark.jars.packages': 'org.apache.spark:spark-streaming-kafka-0-8_2.11:2.3.2,org.apache.kafka:kafka_2.11:1.0.1'}, 'kind': 'pyspark'}</tt><br>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>User</th><th>Current session?</th></tr><tr><td>3065</td><td>application_1618324153128_2643</td><td>pyspark</td><td>error</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2643/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster077.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2643_01_000001/ebouille\">Link</a></td><td>None</td><td></td></tr><tr><td>3086</td><td>application_1618324153128_2664</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2664/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster069.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2664_01_000001/ebouille\">Link</a></td><td>None</td><td></td></tr><tr><td>3091</td><td>application_1618324153128_2669</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2669/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster071.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2669_01_000001/ebouille\">Link</a></td><td>None</td><td></td></tr><tr><td>3092</td><td>application_1618324153128_2670</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2670/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster077.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2670_01_000001/ebouille\">Link</a></td><td>None</td><td></td></tr><tr><td>3093</td><td>application_1618324153128_2671</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2671/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster071.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2671_01_000001/ebouille\">Link</a></td><td>None</td><td></td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%local\n",
    "import os\n",
    "import json\n",
    "\n",
    "username = os.environ['JUPYTERHUB_USER']\n",
    "\n",
    "configuration = dict(\n",
    "    name = \"{}-homework4\".format(username),\n",
    "    executorMemory = \"1G\",\n",
    "    executorCores = 2,\n",
    "    numExecutors = 2,\n",
    "    conf = {\n",
    "        \"spark.jars.packages\":\"org.apache.spark:spark-streaming-kafka-0-8_2.11:2.3.2,org.apache.kafka:kafka_2.11:1.0.1\"\n",
    "    }\n",
    ")\n",
    "\n",
    "ipython = get_ipython()\n",
    "ipython.run_cell_magic('configure', line=\"-f\", cell=json.dumps(configuration))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a new session unless one was already created above (check for `✔` in current session)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>User</th><th>Current session?</th></tr><tr><td>3096</td><td>application_1618324153128_2674</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://iccluster040.iccluster.epfl.ch:8088/proxy/application_1618324153128_2674/\">Link</a></td><td><a target=\"_blank\" href=\"http://iccluster071.iccluster.epfl.ch:8042/node/containerlogs/container_e01_1618324153128_2674_01_000001/ebouille\">Link</a></td><td>None</td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Initialize spark application"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Send `username` to the Spark driver."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Successfully passed 'username' as 'username' to Spark kernel"
     ]
    }
   ],
   "source": [
    "%%send_to_spark -i username -t str -n username"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "You are cadei on the Spark driver."
     ]
    }
   ],
   "source": [
    "print(\"You are {} on the Spark driver.\".format(username))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a Kafka client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pykafka import KafkaClient\n",
    "from pykafka.common import OffsetType\n",
    "\n",
    "ZOOKEEPER_QUORUM = 'iccluster040.iccluster.epfl.ch:2181,' \\\n",
    "                   'iccluster064.iccluster.epfl.ch:2181,' \\\n",
    "                   'iccluster065.iccluster.epfl.ch:2181'\n",
    "\n",
    "client = KafkaClient(zookeeper_hosts=ZOOKEEPER_QUORUM)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Working on data streams is often times more complex compared to using static datasets, so we will first look at how to create static RDDs for easy prototyping.\n",
    "\n",
    "You can find below a function that creates a static RDD from a Kafka topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from itertools import islice\n",
    "\n",
    "def simple_create_rdd(topic, from_offset, to_offset):\n",
    "    \"\"\"Create an RDD from topic with offset in [from_offset, to_offest).\"\"\"\n",
    "    \n",
    "    consumer = client.topics[topic].get_simple_consumer(\n",
    "        auto_offset_reset=OffsetType.EARLIEST if from_offset == 0 else from_offset - 1,\n",
    "        reset_offset_on_start=True\n",
    "    )\n",
    "    \n",
    "    return sc.parallelize((msg.offset, msg.value) for msg in islice(consumer, to_offset - from_offset))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To check this function, we need to retrieve valid offsets from Kafka."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{0: OffsetPartitionResponse(offset=[0], err=0)}"
     ]
    }
   ],
   "source": [
    "topic = client.topics[b'ndovloketnl-arrivals']\n",
    "topic.earliest_available_offsets()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we can for example retrieve the first 1000 messages from the topic `ndovloketnl-arrivals`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "offset = topic.earliest_available_offsets()[0].offset[0]\n",
    "rdd = simple_create_rdd(b'ndovloketnl-arrivals', offset, offset+1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0, '{\"ns1:PutReisInformatieBoodschapIn\": {\"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1\", \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \"ns2:ReisInformatieProductDAS\": {\"@TimeStamp\": \"2021-04-26T11:18:31.081Z\", \"@Versie\": \"6.1\", \"ns2:RIPAdministratie\": {\"ns2:ReisInformatieProductID\": \"2104261318312400002\", \"ns2:AbonnementId\": \"55\", \"ns2:ReisInformatieTijdstip\": \"2021-04-26T11:20:00.000Z\"}, \"ns2:DynamischeAankomstStaat\": {\"ns2:RitId\": \"37542\", \"ns2:RitDatum\": \"2021-04-26\", \"ns2:RitStation\": {\"ns2:StationCode\": \"WS\", \"ns2:Type\": \"1\", \"ns2:KorteNaam\": \"Winschoten\", \"ns2:MiddelNaam\": \"Winschoten\", \"ns2:LangeNaam\": \"Winschoten\", \"ns2:UICCode\": \"8400696\"}, \"ns2:TreinAankomst\": {\"ns2:TreinNummer\": \"37542\", \"ns2:TreinSoort\": {\"@Code\": \"ST\", \"#text\": \"Stoptrein\"}, \"ns2:TreinStatus\": \"2\", \"ns2:LijnNummer\": \"RS6\", \"ns2:Vervoerder\": \"Arriva\", \"ns2:TreinHerkomst\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"GN\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Groningen\", \"ns2:MiddelNaam\": \"Groningen\", \"ns2:LangeNaam\": \"Groningen\", \"ns2:UICCode\": \"8400263\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"GN\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Groningen\", \"ns2:MiddelNaam\": \"Groningen\", \"ns2:LangeNaam\": \"Groningen\", \"ns2:UICCode\": \"8400263\"}], \"ns2:PresentatieTreinHerkomst\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Groningen\"}}, \"ns2:AankomstTijd\": [{\"@InfoStatus\": \"Gepland\", \"#text\": \"2021-04-26T11:20:00.000Z\"}, {\"@InfoStatus\": \"Actueel\", \"#text\": \"2021-04-26T11:20:00.000Z\"}], \"ns2:ExacteAankomstVertraging\": \"PT0S\", \"ns2:GedempteAankomstVertraging\": \"PT0S\", \"ns2:TreinAankomstSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"1\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"1\"}], \"ns2:PresentatieTreinAankomstSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"1\"}}, \"ns2:VerkorteRouteHerkomst\": [{\"@InfoStatus\": \"Gepland\", \"ns2:Station\": [{\"ns2:StationCode\": \"HGZ\", \"ns2:Type\": \"1\", \"ns2:KorteNaam\": \"Hoogezand\", \"ns2:MiddelNaam\": \"Hoogezand-Sap.\", \"ns2:LangeNaam\": \"Hoogezand-Sappemeer\", \"ns2:UICCode\": \"8400331\"}, {\"ns2:StationCode\": \"ZB\", \"ns2:Type\": \"1\", \"ns2:KorteNaam\": \"Zuidbroek\", \"ns2:MiddelNaam\": \"Zuidbroek\", \"ns2:LangeNaam\": \"Zuidbroek\", \"ns2:UICCode\": \"8400742\"}]}, {\"@InfoStatus\": \"Actueel\", \"ns2:Station\": [{\"ns2:StationCode\": \"HGZ\", \"ns2:Type\": \"1\", \"ns2:KorteNaam\": \"Hoogezand\", \"ns2:MiddelNaam\": \"Hoogezand-Sap.\", \"ns2:LangeNaam\": \"Hoogezand-Sappemeer\", \"ns2:UICCode\": \"8400331\"}, {\"ns2:StationCode\": \"ZB\", \"ns2:Type\": \"1\", \"ns2:KorteNaam\": \"Zuidbroek\", \"ns2:MiddelNaam\": \"Zuidbroek\", \"ns2:LangeNaam\": \"Zuidbroek\", \"ns2:UICCode\": \"8400742\"}]}], \"ns2:PresentatieVerkorteRouteHerkomst\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Hoogezand-Sap., Zuidbroek\"}}, \"ns2:WijzigingHerkomst\": {\"ns2:WijzigingType\": \"40\"}}}}}}')"
     ]
    }
   ],
   "source": [
    "rdd.first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1000"
     ]
    }
   ],
   "source": [
    "rdd.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Streams from Kafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "checkpoint created at hdfs:///user/cadei/checkpoint/"
     ]
    }
   ],
   "source": [
    "# Define the checkpoint folder\n",
    "checkpoint = 'hdfs:///user/{}/checkpoint/'.format(username)\n",
    "print('checkpoint created at hdfs:///user/{}/checkpoint/'.format(username))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "\n",
    "# Create a StreamingContext with two working thread and batch interval of 5 seconds.\n",
    "# Each time you stop a StreamingContext, you will need to recreate it.\n",
    "ssc = StreamingContext(sc, 10)\n",
    "ssc.checkpoint(checkpoint)\n",
    "\n",
    "group_id = 'ns-{0}'.format(username)\n",
    "\n",
    "# Input streams\n",
    "arrival_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-arrivals': 1})\n",
    "departure_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-departures': 1})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For now, let's just print the content of the streams. Because we set the batch interval as 10 seconds and the timeout also as 10 seconds, you are supposed to see exact one batch from each stream, like:\n",
    "```\n",
    "-------------------------------------------\n",
    "Time: 2021-04-27 10:11:50\n",
    "-------------------------------------------\n",
    "<ONE_BATCH_OF_ARRIVAL_STREAM>\n",
    "...\n",
    "-------------------------------------------\n",
    "Time: 2021-04-27 10:11:50\n",
    "-------------------------------------------\n",
    "<ONE_BATCH_OF_DEPARTURE_STREAM>\n",
    "...\n",
    "```\n",
    "**Note:** the output may be shown after you run `ssc.stop`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:39:40\n",
      "-------------------------------------------\n",
      "(None, u'{\"ns1:PutReisInformatieBoodschapIn\": {\"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1\", \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \"ns2:ReisInformatieProductDAS\": {\"@TimeStamp\": \"2021-05-11T20:39:33.331Z\", \"@Versie\": \"6.1\", \"ns2:RIPAdministratie\": {\"ns2:ReisInformatieProductID\": \"2105112239332400004\", \"ns2:AbonnementId\": \"55\", \"ns2:ReisInformatieTijdstip\": \"2021-05-11T20:40:00.000Z\"}, \"ns2:DynamischeAankomstStaat\": {\"ns2:RitId\": \"4380\", \"ns2:RitDatum\": \"2021-05-11\", \"ns2:RitStation\": {\"ns2:StationCode\": \"ALMB\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Buiten\", \"ns2:MiddelNaam\": \"Buiten\", \"ns2:LangeNaam\": \"Almere Buiten\", \"ns2:UICCode\": \"8400081\"}, \"ns2:TreinAankomst\": {\"ns2:TreinNummer\": \"4380\", \"ns2:TreinSoort\": {\"@Code\": \"SPR\", \"#text\": \"Sprinter\"}, \"ns2:TreinStatus\": \"2\", \"ns2:Vervoerder\": \"NS\", \"ns2:TreinHerkomst\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ALMO\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Oostvaard\", \"ns2:MiddelNaam\": \"Oostvaarders\", \"ns2:LangeNaam\": \"Almere Oostvaarders\", \"ns2:UICCode\": \"8400226\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ALMO\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Oostvaard\", \"ns2:MiddelNaam\": \"Oostvaarders\", \"ns2:LangeNaam\": \"Almere Oostvaarders\", \"ns2:UICCode\": \"8400226\"}], \"ns2:PresentatieTreinHerkomst\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Oostvaarders\"}}, \"ns2:AankomstTijd\": [{\"@InfoStatus\": \"Gepland\", \"#text\": \"2021-05-11T20:40:00.000Z\"}, {\"@InfoStatus\": \"Actueel\", \"#text\": \"2021-05-11T20:40:00.000Z\"}], \"ns2:ExacteAankomstVertraging\": \"PT0S\", \"ns2:GedempteAankomstVertraging\": \"PT0S\", \"ns2:TreinAankomstSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"1\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"1\"}], \"ns2:PresentatieTreinAankomstSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"1\"}}, \"ns2:WijzigingHerkomst\": {\"ns2:WijzigingType\": \"40\"}}}}}}')\n",
      "(None, u'{\"ns1:PutReisInformatieBoodschapIn\": {\"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1\", \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \"ns2:ReisInformatieProductDAS\": {\"@TimeStamp\": \"2021-05-11T20:39:34.334Z\", \"@Versie\": \"6.1\", \"ns2:RIPAdministratie\": {\"ns2:ReisInformatieProductID\": \"2105112239342400002\", \"ns2:AbonnementId\": \"55\", \"ns2:ReisInformatieTijdstip\": \"2021-05-11T20:39:00.000Z\"}, \"ns2:DynamischeAankomstStaat\": {\"ns2:RitId\": \"31284\", \"ns2:RitDatum\": \"2021-05-11\", \"ns2:RitStation\": {\"ns2:StationCode\": \"LC\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Lochem\", \"ns2:MiddelNaam\": \"Lochem\", \"ns2:LangeNaam\": \"Lochem\", \"ns2:UICCode\": \"8400399\"}, \"ns2:TreinAankomst\": {\"ns2:TreinNummer\": \"31284\", \"ns2:TreinSoort\": {\"@Code\": \"ST\", \"#text\": \"Stoptrein\"}, \"ns2:TreinStatus\": \"5\", \"ns2:Vervoerder\": \"Blauwnet\", \"ns2:TreinHerkomst\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ODZ\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Oldenzaal\", \"ns2:MiddelNaam\": \"Oldenzaal\", \"ns2:LangeNaam\": \"Oldenzaal\", \"ns2:UICCode\": \"8400483\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ODZ\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Oldenzaal\", \"ns2:MiddelNaam\": \"Oldenzaal\", \"ns2:LangeNaam\": \"Oldenzaal\", \"ns2:UICCode\": \"8400483\"}], \"ns2:PresentatieTreinHerkomst\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Oldenzaal\"}}, \"ns2:AankomstTijd\": [{\"@InfoStatus\": \"Gepland\", \"#text\": \"2021-05-11T20:39:00.000Z\"}, {\"@InfoStatus\": \"Actueel\", \"#text\": \"2021-05-11T20:39:00.000Z\"}], \"ns2:ExacteAankomstVertraging\": \"PT0S\", \"ns2:GedempteAankomstVertraging\": \"PT0S\", \"ns2:TreinAankomstSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"1\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"1\"}], \"ns2:PresentatieTreinAankomstSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"1\"}}, \"ns2:VerkorteRouteHerkomst\": [{\"@InfoStatus\": \"Gepland\", \"ns2:Station\": {\"ns2:StationCode\": \"HGL\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Hengelo\", \"ns2:MiddelNaam\": \"Hengelo\", \"ns2:LangeNaam\": \"Hengelo\", \"ns2:UICCode\": \"8400316\"}}, {\"@InfoStatus\": \"Actueel\", \"ns2:Station\": {\"ns2:StationCode\": \"HGL\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Hengelo\", \"ns2:MiddelNaam\": \"Hengelo\", \"ns2:LangeNaam\": \"Hengelo\", \"ns2:UICCode\": \"8400316\"}}], \"ns2:PresentatieVerkorteRouteHerkomst\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Hengelo\"}}, \"ns2:WijzigingHerkomst\": {\"ns2:WijzigingType\": \"40\"}}}}}}')\n",
      "...\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:39:40\n",
      "-------------------------------------------\n",
      "(None, u'{\"ns1:PutReisInformatieBoodschapIn\": {\"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:5\", \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \"ns2:ReisInformatieProductDVS\": {\"@TimeStamp\": \"2021-05-11T20:39:34.328Z\", \"@Versie\": \"6.2\", \"ns2:RIPAdministratie\": {\"ns2:ReisInformatieProductID\": \"2105112239342400001\", \"ns2:AbonnementId\": \"54\", \"ns2:ReisInformatieTijdstip\": \"2021-05-11T20:39:00.000Z\"}, \"ns2:DynamischeVertrekStaat\": {\"ns2:RitId\": \"31284\", \"ns2:RitDatum\": \"2021-05-11\", \"ns2:RitStation\": {\"ns2:StationCode\": \"LC\", \"ns2:Type\": \"0\", \"ns2:KorteNaam\": \"Lochem\", \"ns2:MiddelNaam\": \"Lochem\", \"ns2:LangeNaam\": \"Lochem\", \"ns2:UICCode\": \"8400399\"}, \"ns2:Trein\": {\"ns2:TreinNummer\": \"31284\", \"ns2:TreinSoort\": {\"@Code\": \"ST\", \"#text\": \"Stoptrein\"}, \"ns2:TreinFormule\": \"1\", \"ns2:TreinStatus\": \"5\", \"ns2:Vervoerder\": \"Blauwnet\", \"ns2:Reserveren\": \"N\", \"ns2:Toeslag\": \"N\", \"ns2:NietInstappen\": \"N\", \"ns2:AchterBlijvenAchtersteTreinDeel\": \"N\", \"ns2:RangeerBeweging\": \"N\", \"ns2:SpeciaalKaartje\": \"N\", \"ns2:TreinEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}], \"ns2:PresentatieTreinEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Zutphen\"}}, \"ns2:VertrekTijd\": [{\"@InfoStatus\": \"Gepland\", \"#text\": \"2021-05-11T20:39:00.000Z\"}, {\"@InfoStatus\": \"Actueel\", \"#text\": \"2021-05-11T20:39:00.000Z\"}], \"ns2:ExacteVertrekVertraging\": \"PT0S\", \"ns2:GedempteVertrekVertraging\": \"PT0S\", \"ns2:TreinVertrekSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"1\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"1\"}], \"ns2:PresentatieTreinVertrekSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"1\"}}, \"ns2:VertrekRichting\": \"A\", \"ns2:AfstandPerronEindKopVertrekTrein\": \"0\", \"ns2:TreinVleugel\": {\"ns2:TreinVleugelVertrekSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"1\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"1\"}], \"ns2:PresentatieTreinVleugelVertrekSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"1\"}}, \"ns2:TreinVleugelEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}], \"ns2:PresentatieTreinVleugelEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Zutphen\"}}, \"ns2:StopStations\": [{\"@InfoStatus\": \"Gepland\", \"ns2:Station\": {\"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}}, {\"@InfoStatus\": \"Actueel\", \"ns2:Station\": {\"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}}], \"ns2:MaterieelDeelDVS\": {\"ns2:MaterieelSoort\": \"LINT KE\", \"ns2:MaterieelAanduiding\": \"2\", \"ns2:MaterieelLengte\": \"4100\", \"ns2:MaterieelDeelEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ZP\", \"ns2:Type\": \"5\", \"ns2:KorteNaam\": \"Zutphen\", \"ns2:MiddelNaam\": \"Zutphen\", \"ns2:LangeNaam\": \"Zutphen\", \"ns2:UICCode\": \"8400744\"}], \"ns2:PresentatieMaterieelDeelEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Zutphen\"}}}}, \"ns2:Wijziging\": {\"ns2:WijzigingType\": \"40\"}}}}}}')\n",
      "(None, u'{\"ns1:PutReisInformatieBoodschapIn\": {\"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:5\", \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \"ns2:ReisInformatieProductDVS\": {\"@TimeStamp\": \"2021-05-11T20:39:32.181Z\", \"@Versie\": \"6.2\", \"ns2:RIPAdministratie\": {\"ns2:ReisInformatieProductID\": \"2105112239322500002\", \"ns2:AbonnementId\": \"54\", \"ns2:ReisInformatieTijdstip\": \"2021-05-11T20:40:00.000Z\"}, \"ns2:DynamischeVertrekStaat\": {\"ns2:RitId\": \"967\", \"ns2:RitDatum\": \"2021-05-11\", \"ns2:RitStation\": {\"ns2:StationCode\": \"RTD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Rotterdm C\", \"ns2:MiddelNaam\": \"Rotterdam C.\", \"ns2:LangeNaam\": \"Rotterdam Centraal\", \"ns2:UICCode\": \"8400530\"}, \"ns2:Trein\": {\"ns2:TreinNummer\": \"967\", \"ns2:TreinNaam\": \"Toeslag/suppl. Schiphol-R\\'dam\", \"ns2:TreinSoort\": {\"@Code\": \"ICD\", \"#text\": \"Intercity direct\"}, \"ns2:TreinFormule\": \"5\", \"ns2:TreinStatus\": \"0\", \"ns2:Vervoerder\": \"NS\", \"ns2:Reserveren\": \"N\", \"ns2:Toeslag\": \"N\", \"ns2:NietInstappen\": \"N\", \"ns2:AchterBlijvenAchtersteTreinDeel\": \"N\", \"ns2:RangeerBeweging\": \"N\", \"ns2:SpeciaalKaartje\": \"N\", \"ns2:TreinEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}], \"ns2:PresentatieTreinEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Amsterdam Centraal\"}}, \"ns2:VertrekTijd\": [{\"@InfoStatus\": \"Gepland\", \"#text\": \"2021-05-11T20:40:00.000Z\"}, {\"@InfoStatus\": \"Actueel\", \"#text\": \"2021-05-11T20:44:19.000Z\"}], \"ns2:ExacteVertrekVertraging\": \"PT4M19S\", \"ns2:PresentatieVertrekVertraging\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"+4 min.\"}}, \"ns2:GedempteVertrekVertraging\": \"PT5M\", \"ns2:TreinVertrekSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"12\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"12\"}], \"ns2:PresentatieTreinVertrekSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"12\"}}, \"ns2:VertrekRichting\": \"A\", \"ns2:AfstandPerronEindKopVertrekTrein\": \"0\", \"ns2:VerkorteRoute\": [{\"@InfoStatus\": \"Gepland\", \"ns2:Station\": {\"ns2:StationCode\": \"SHL\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Schiphol\", \"ns2:MiddelNaam\": \"Schiphol Airport\", \"ns2:LangeNaam\": \"Schiphol Airport\", \"ns2:UICCode\": \"8400561\"}}, {\"@InfoStatus\": \"Actueel\", \"ns2:Station\": {\"ns2:StationCode\": \"SHL\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Schiphol\", \"ns2:MiddelNaam\": \"Schiphol Airport\", \"ns2:LangeNaam\": \"Schiphol Airport\", \"ns2:UICCode\": \"8400561\"}}], \"ns2:PresentatieVerkorteRoute\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Schiphol Airport\"}}, \"ns2:TreinVleugel\": {\"ns2:TreinVleugelVertrekSpoor\": [{\"@InfoStatus\": \"Gepland\", \"ns2:SpoorNummer\": \"12\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:SpoorNummer\": \"12\"}], \"ns2:PresentatieTreinVleugelVertrekSpoor\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"12\"}}, \"ns2:TreinVleugelEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}], \"ns2:PresentatieTreinVleugelEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Amsterdam Centraal\"}}, \"ns2:StopStations\": [{\"@InfoStatus\": \"Gepland\", \"ns2:Station\": [{\"ns2:StationCode\": \"SHL\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Schiphol\", \"ns2:MiddelNaam\": \"Schiphol Airport\", \"ns2:LangeNaam\": \"Schiphol Airport\", \"ns2:UICCode\": \"8400561\"}, {\"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}]}, {\"@InfoStatus\": \"Actueel\", \"ns2:Station\": [{\"ns2:StationCode\": \"SHL\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Schiphol\", \"ns2:MiddelNaam\": \"Schiphol Airport\", \"ns2:LangeNaam\": \"Schiphol Airport\", \"ns2:UICCode\": \"8400561\"}, {\"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}]}], \"ns2:MaterieelDeelDVS\": {\"ns2:MaterieelSoort\": \"SW7-25KV\", \"ns2:MaterieelAanduiding\": \"2+7\", \"ns2:MaterieelLengte\": \"22200\", \"ns2:MaterieelNummer\": \"000000-16125-0\", \"ns2:MaterieelDeelEindBestemming\": [{\"@InfoStatus\": \"Gepland\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}, {\"@InfoStatus\": \"Actueel\", \"ns2:StationCode\": \"ASD\", \"ns2:Type\": \"6\", \"ns2:KorteNaam\": \"Amsterdm C\", \"ns2:MiddelNaam\": \"Amsterdam C.\", \"ns2:LangeNaam\": \"Amsterdam Centraal\", \"ns2:UICCode\": \"8400058\"}], \"ns2:PresentatieMaterieelDeelEindBestemming\": {\"ns2:Uitingen\": {\"ns2:Uiting\": \"Amsterdam Centraal\"}}}}, \"ns2:Wijziging\": {\"ns2:WijzigingType\": \"10\", \"ns2:PresentatieWijziging\": {\"ns2:Uitingen\": [{\"@Taal\": \"nl\", \"ns2:Uiting\": {\"@Prioriteit\": \"10\", \"#text\": \"Later vertrek\"}}, {\"@Taal\": \"en\", \"ns2:Uiting\": {\"@Prioriteit\": \"10\", \"#text\": \"Delayed\"}}]}}}, \"ns2:PresentatieOpmerkingen\": {\"ns2:Uitingen\": [{\"@Taal\": \"nl\", \"ns2:Uiting\": {\"@Prioriteit\": \"20\", \"@ReferentieType\": \"Wijziging\", \"@ReferentieWaarde\": \"10\", \"#text\": \"Later vertrek\"}}, {\"@Taal\": \"en\", \"ns2:Uiting\": {\"@Prioriteit\": \"20\", \"@ReferentieType\": \"Wijziging\", \"@ReferentieWaarde\": \"10\", \"#text\": \"Delayed\"}}]}}}}}')\n",
      "..."
     ]
    }
   ],
   "source": [
    "arrival_stream.pprint(num=2) # print the first 2 messages\n",
    "departure_stream.pprint(num=2) # print the first 2 messages\n",
    "\n",
    "ssc.start()\n",
    "ssc.awaitTermination(timeout=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "ssc.stop(stopSparkContext=False, stopGraceFully=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You will need to adjust the batch interval (10 seconds here) in accordance with the processing times. Use the spark UI to check if batches are not accumulating."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Part III - Live stopping time (20 points)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this part, we will have a look at the two other streams, namely `ndovloketnl-arrivals` and `ndovloketnl-departures`. Each time a train arrives at or leaves a station, a message is generated. Let's have a look at the content."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "  \"ns1:PutReisInformatieBoodschapIn\": {\n",
      "    \"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1\", \n",
      "    \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\", \n",
      "    \"ns2:ReisInformatieProductDAS\": {\n",
      "      \"ns2:RIPAdministratie\": {\n",
      "        \"ns2:ReisInformatieTijdstip\": \"2021-04-26T11:20:00.000Z\", \n",
      "        \"ns2:ReisInformatieProductID\": \"2104261318312400002\", \n",
      "        \"ns2:AbonnementId\": \"55\"\n",
      "      }, \n",
      "      \"@TimeStamp\": \"2021-04-26T11:18:31.081Z\", \n",
      "      \"@Versie\": \"6.1\", \n",
      "      \"ns2:DynamischeAankomstStaat\": {\n",
      "        \"ns2:RitId\": \"37542\", \n",
      "        \"ns2:RitStation\": {\n",
      "          \"ns2:LangeNaam\": \"Winschoten\", \n",
      "          \"ns2:MiddelNaam\": \"Winschoten\", \n",
      "          \"ns2:StationCode\": \"WS\", \n",
      "          \"ns2:Type\": \"1\", \n",
      "          \"ns2:UICCode\": \"8400696\", \n",
      "          \"ns2:KorteNaam\": \"Winschoten\"\n",
      "        }, \n",
      "        \"ns2:TreinAankomst\": {\n",
      "          \"ns2:TreinHerkomst\": [\n",
      "            {\n",
      "              \"ns2:LangeNaam\": \"Groningen\", \n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"ns2:MiddelNaam\": \"Groningen\", \n",
      "              \"ns2:StationCode\": \"GN\", \n",
      "              \"ns2:Type\": \"5\", \n",
      "              \"ns2:UICCode\": \"8400263\", \n",
      "              \"ns2:KorteNaam\": \"Groningen\"\n",
      "            }, \n",
      "            {\n",
      "              \"ns2:LangeNaam\": \"Groningen\", \n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"ns2:MiddelNaam\": \"Groningen\", \n",
      "              \"ns2:StationCode\": \"GN\", \n",
      "              \"ns2:Type\": \"5\", \n",
      "              \"ns2:UICCode\": \"8400263\", \n",
      "              \"ns2:KorteNaam\": \"Groningen\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:TreinSoort\": {\n",
      "            \"@Code\": \"ST\", \n",
      "            \"#text\": \"Stoptrein\"\n",
      "          }, \n",
      "          \"ns2:WijzigingHerkomst\": {\n",
      "            \"ns2:WijzigingType\": \"40\"\n",
      "          }, \n",
      "          \"ns2:VerkorteRouteHerkomst\": [\n",
      "            {\n",
      "              \"ns2:Station\": [\n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Hoogezand-Sappemeer\", \n",
      "                  \"ns2:MiddelNaam\": \"Hoogezand-Sap.\", \n",
      "                  \"ns2:StationCode\": \"HGZ\", \n",
      "                  \"ns2:Type\": \"1\", \n",
      "                  \"ns2:UICCode\": \"8400331\", \n",
      "                  \"ns2:KorteNaam\": \"Hoogezand\"\n",
      "                }, \n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Zuidbroek\", \n",
      "                  \"ns2:MiddelNaam\": \"Zuidbroek\", \n",
      "                  \"ns2:StationCode\": \"ZB\", \n",
      "                  \"ns2:Type\": \"1\", \n",
      "                  \"ns2:UICCode\": \"8400742\", \n",
      "                  \"ns2:KorteNaam\": \"Zuidbroek\"\n",
      "                }\n",
      "              ], \n",
      "              \"@InfoStatus\": \"Gepland\"\n",
      "            }, \n",
      "            {\n",
      "              \"ns2:Station\": [\n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Hoogezand-Sappemeer\", \n",
      "                  \"ns2:MiddelNaam\": \"Hoogezand-Sap.\", \n",
      "                  \"ns2:StationCode\": \"HGZ\", \n",
      "                  \"ns2:Type\": \"1\", \n",
      "                  \"ns2:UICCode\": \"8400331\", \n",
      "                  \"ns2:KorteNaam\": \"Hoogezand\"\n",
      "                }, \n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Zuidbroek\", \n",
      "                  \"ns2:MiddelNaam\": \"Zuidbroek\", \n",
      "                  \"ns2:StationCode\": \"ZB\", \n",
      "                  \"ns2:Type\": \"1\", \n",
      "                  \"ns2:UICCode\": \"8400742\", \n",
      "                  \"ns2:KorteNaam\": \"Zuidbroek\"\n",
      "                }\n",
      "              ], \n",
      "              \"@InfoStatus\": \"Actueel\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:LijnNummer\": \"RS6\", \n",
      "          \"ns2:PresentatieTreinAankomstSpoor\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"1\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:TreinStatus\": \"2\", \n",
      "          \"ns2:PresentatieVerkorteRouteHerkomst\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"Hoogezand-Sap., Zuidbroek\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:AankomstTijd\": [\n",
      "            {\n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"#text\": \"2021-04-26T11:20:00.000Z\"\n",
      "            }, \n",
      "            {\n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"#text\": \"2021-04-26T11:20:00.000Z\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:TreinNummer\": \"37542\", \n",
      "          \"ns2:TreinAankomstSpoor\": [\n",
      "            {\n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"ns2:SpoorNummer\": \"1\"\n",
      "            }, \n",
      "            {\n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"ns2:SpoorNummer\": \"1\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:PresentatieTreinHerkomst\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"Groningen\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:GedempteAankomstVertraging\": \"PT0S\", \n",
      "          \"ns2:ExacteAankomstVertraging\": \"PT0S\", \n",
      "          \"ns2:Vervoerder\": \"Arriva\"\n",
      "        }, \n",
      "        \"ns2:RitDatum\": \"2021-04-26\"\n",
      "      }\n",
      "    }\n",
      "  }\n",
      "}"
     ]
    }
   ],
   "source": [
    "import json\n",
    "from pykafka.common import OffsetType\n",
    "\n",
    "example_arrivals = client.topics[b'ndovloketnl-arrivals'].get_simple_consumer(\n",
    "    auto_offset_reset=OffsetType.EARLIEST,\n",
    "    reset_offset_on_start=True\n",
    ").consume()\n",
    "print(json.dumps(json.loads(example_arrivals.value), indent=2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "  \"ns1:PutReisInformatieBoodschapIn\": {\n",
      "    \"@xmlns:ns1\": \"urn:ndov:cdm:trein:reisinformatie:messages:5\", \n",
      "    \"ns2:ReisInformatieProductDVS\": {\n",
      "      \"ns2:RIPAdministratie\": {\n",
      "        \"ns2:ReisInformatieTijdstip\": \"2021-04-28T12:07:00.000Z\", \n",
      "        \"ns2:ReisInformatieProductID\": \"2104281408062300003\", \n",
      "        \"ns2:AbonnementId\": \"54\"\n",
      "      }, \n",
      "      \"@TimeStamp\": \"2021-04-28T12:08:06.847Z\", \n",
      "      \"@Versie\": \"6.2\", \n",
      "      \"ns2:DynamischeVertrekStaat\": {\n",
      "        \"ns2:RitId\": \"31241\", \n",
      "        \"ns2:RitStation\": {\n",
      "          \"ns2:LangeNaam\": \"Delden\", \n",
      "          \"ns2:MiddelNaam\": \"Delden\", \n",
      "          \"ns2:StationCode\": \"DDN\", \n",
      "          \"ns2:Type\": \"0\", \n",
      "          \"ns2:UICCode\": \"8400169\", \n",
      "          \"ns2:KorteNaam\": \"Delden\"\n",
      "        }, \n",
      "        \"ns2:RitDatum\": \"2021-04-28\", \n",
      "        \"ns2:Trein\": {\n",
      "          \"ns2:VertrekRichting\": \"B\", \n",
      "          \"ns2:VertrekTijd\": [\n",
      "            {\n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"#text\": \"2021-04-28T12:07:00.000Z\"\n",
      "            }, \n",
      "            {\n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"#text\": \"2021-04-28T12:07:00.000Z\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:RangeerBeweging\": \"N\", \n",
      "          \"ns2:PresentatieTreinVertrekSpoor\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"1\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:NietInstappen\": \"N\", \n",
      "          \"ns2:PresentatieVerkorteRoute\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"Hengelo\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:TreinStatus\": \"5\", \n",
      "          \"ns2:TreinNummer\": \"31241\", \n",
      "          \"ns2:ExacteVertrekVertraging\": \"PT0S\", \n",
      "          \"ns2:Wijziging\": {\n",
      "            \"ns2:WijzigingType\": \"40\"\n",
      "          }, \n",
      "          \"ns2:TreinVleugel\": {\n",
      "            \"ns2:PresentatieTreinVleugelVertrekSpoor\": {\n",
      "              \"ns2:Uitingen\": {\n",
      "                \"ns2:Uiting\": \"1\"\n",
      "              }\n",
      "            }, \n",
      "            \"ns2:TreinVleugelVertrekSpoor\": [\n",
      "              {\n",
      "                \"@InfoStatus\": \"Gepland\", \n",
      "                \"ns2:SpoorNummer\": \"1\"\n",
      "              }, \n",
      "              {\n",
      "                \"@InfoStatus\": \"Actueel\", \n",
      "                \"ns2:SpoorNummer\": \"1\"\n",
      "              }\n",
      "            ], \n",
      "            \"ns2:StopStations\": [\n",
      "              {\n",
      "                \"ns2:Station\": [\n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo Gezondheidspark\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo Gezondh.\", \n",
      "                    \"ns2:StationCode\": \"HGLG\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400433\", \n",
      "                    \"ns2:KorteNaam\": \"Gezondhprk\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo\", \n",
      "                    \"ns2:StationCode\": \"HGL\", \n",
      "                    \"ns2:Type\": \"5\", \n",
      "                    \"ns2:UICCode\": \"8400316\", \n",
      "                    \"ns2:KorteNaam\": \"Hengelo\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo Oost\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo O.\", \n",
      "                    \"ns2:StationCode\": \"HGLO\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400312\", \n",
      "                    \"ns2:KorteNaam\": \"Hengelo O\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                    \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                    \"ns2:StationCode\": \"ODZ\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400483\", \n",
      "                    \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "                  }\n",
      "                ], \n",
      "                \"@InfoStatus\": \"Gepland\"\n",
      "              }, \n",
      "              {\n",
      "                \"ns2:Station\": [\n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo Gezondheidspark\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo Gezondh.\", \n",
      "                    \"ns2:StationCode\": \"HGLG\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400433\", \n",
      "                    \"ns2:KorteNaam\": \"Gezondhprk\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo\", \n",
      "                    \"ns2:StationCode\": \"HGL\", \n",
      "                    \"ns2:Type\": \"5\", \n",
      "                    \"ns2:UICCode\": \"8400316\", \n",
      "                    \"ns2:KorteNaam\": \"Hengelo\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Hengelo Oost\", \n",
      "                    \"ns2:MiddelNaam\": \"Hengelo O.\", \n",
      "                    \"ns2:StationCode\": \"HGLO\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400312\", \n",
      "                    \"ns2:KorteNaam\": \"Hengelo O\"\n",
      "                  }, \n",
      "                  {\n",
      "                    \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                    \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                    \"ns2:StationCode\": \"ODZ\", \n",
      "                    \"ns2:Type\": \"0\", \n",
      "                    \"ns2:UICCode\": \"8400483\", \n",
      "                    \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "                  }\n",
      "                ], \n",
      "                \"@InfoStatus\": \"Actueel\"\n",
      "              }\n",
      "            ], \n",
      "            \"ns2:TreinVleugelEindBestemming\": [\n",
      "              {\n",
      "                \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                \"@InfoStatus\": \"Gepland\", \n",
      "                \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                \"ns2:StationCode\": \"ODZ\", \n",
      "                \"ns2:Type\": \"0\", \n",
      "                \"ns2:UICCode\": \"8400483\", \n",
      "                \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "              }, \n",
      "              {\n",
      "                \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                \"@InfoStatus\": \"Actueel\", \n",
      "                \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                \"ns2:StationCode\": \"ODZ\", \n",
      "                \"ns2:Type\": \"0\", \n",
      "                \"ns2:UICCode\": \"8400483\", \n",
      "                \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "              }\n",
      "            ], \n",
      "            \"ns2:MaterieelDeelDVS\": {\n",
      "              \"ns2:PresentatieMaterieelDeelEindBestemming\": {\n",
      "                \"ns2:Uitingen\": {\n",
      "                  \"ns2:Uiting\": \"Oldenzaal\"\n",
      "                }\n",
      "              }, \n",
      "              \"ns2:MaterieelLengte\": \"4100\", \n",
      "              \"ns2:MaterieelAanduiding\": \"2\", \n",
      "              \"ns2:MaterieelSoort\": \"LINT KE\", \n",
      "              \"ns2:MaterieelDeelEindBestemming\": [\n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                  \"@InfoStatus\": \"Gepland\", \n",
      "                  \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                  \"ns2:StationCode\": \"ODZ\", \n",
      "                  \"ns2:Type\": \"0\", \n",
      "                  \"ns2:UICCode\": \"8400483\", \n",
      "                  \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "                }, \n",
      "                {\n",
      "                  \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "                  \"@InfoStatus\": \"Actueel\", \n",
      "                  \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "                  \"ns2:StationCode\": \"ODZ\", \n",
      "                  \"ns2:Type\": \"0\", \n",
      "                  \"ns2:UICCode\": \"8400483\", \n",
      "                  \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "                }\n",
      "              ]\n",
      "            }, \n",
      "            \"ns2:PresentatieTreinVleugelEindBestemming\": {\n",
      "              \"ns2:Uitingen\": {\n",
      "                \"ns2:Uiting\": \"Oldenzaal\"\n",
      "              }\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:TreinEindBestemming\": [\n",
      "            {\n",
      "              \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "              \"ns2:StationCode\": \"ODZ\", \n",
      "              \"ns2:Type\": \"0\", \n",
      "              \"ns2:UICCode\": \"8400483\", \n",
      "              \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "            }, \n",
      "            {\n",
      "              \"ns2:LangeNaam\": \"Oldenzaal\", \n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"ns2:MiddelNaam\": \"Oldenzaal\", \n",
      "              \"ns2:StationCode\": \"ODZ\", \n",
      "              \"ns2:Type\": \"0\", \n",
      "              \"ns2:UICCode\": \"8400483\", \n",
      "              \"ns2:KorteNaam\": \"Oldenzaal\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:AchterBlijvenAchtersteTreinDeel\": \"N\", \n",
      "          \"ns2:Reserveren\": \"N\", \n",
      "          \"ns2:AfstandPerronEindKopVertrekTrein\": \"0\", \n",
      "          \"ns2:VerkorteRoute\": [\n",
      "            {\n",
      "              \"ns2:Station\": {\n",
      "                \"ns2:LangeNaam\": \"Hengelo\", \n",
      "                \"ns2:MiddelNaam\": \"Hengelo\", \n",
      "                \"ns2:StationCode\": \"HGL\", \n",
      "                \"ns2:Type\": \"5\", \n",
      "                \"ns2:UICCode\": \"8400316\", \n",
      "                \"ns2:KorteNaam\": \"Hengelo\"\n",
      "              }, \n",
      "              \"@InfoStatus\": \"Gepland\"\n",
      "            }, \n",
      "            {\n",
      "              \"ns2:Station\": {\n",
      "                \"ns2:LangeNaam\": \"Hengelo\", \n",
      "                \"ns2:MiddelNaam\": \"Hengelo\", \n",
      "                \"ns2:StationCode\": \"HGL\", \n",
      "                \"ns2:Type\": \"5\", \n",
      "                \"ns2:UICCode\": \"8400316\", \n",
      "                \"ns2:KorteNaam\": \"Hengelo\"\n",
      "              }, \n",
      "              \"@InfoStatus\": \"Actueel\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:TreinVertrekSpoor\": [\n",
      "            {\n",
      "              \"@InfoStatus\": \"Gepland\", \n",
      "              \"ns2:SpoorNummer\": \"1\"\n",
      "            }, \n",
      "            {\n",
      "              \"@InfoStatus\": \"Actueel\", \n",
      "              \"ns2:SpoorNummer\": \"1\"\n",
      "            }\n",
      "          ], \n",
      "          \"ns2:Toeslag\": \"N\", \n",
      "          \"ns2:TreinSoort\": {\n",
      "            \"@Code\": \"ST\", \n",
      "            \"#text\": \"Stoptrein\"\n",
      "          }, \n",
      "          \"ns2:TreinFormule\": \"1\", \n",
      "          \"ns2:PresentatieTreinEindBestemming\": {\n",
      "            \"ns2:Uitingen\": {\n",
      "              \"ns2:Uiting\": \"Oldenzaal\"\n",
      "            }\n",
      "          }, \n",
      "          \"ns2:SpeciaalKaartje\": \"N\", \n",
      "          \"ns2:Vervoerder\": \"Blauwnet\", \n",
      "          \"ns2:GedempteVertrekVertraging\": \"PT0S\"\n",
      "        }\n",
      "      }\n",
      "    }, \n",
      "    \"@xmlns:ns2\": \"urn:ndov:cdm:trein:reisinformatie:data:4\"\n",
      "  }\n",
      "}"
     ]
    }
   ],
   "source": [
    "example_departures = client.topics[b'ndovloketnl-departures'].get_simple_consumer(\n",
    "    auto_offset_reset=OffsetType.EARLIEST,\n",
    "    reset_offset_on_start=True\n",
    ").consume()\n",
    "print(json.dumps(json.loads(example_departures.value), indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see that the messages have the following structure:\n",
    "\n",
    "```\n",
    "{\n",
    "  'ns1:PutReisInformatieBoodschapIn': {\n",
    "    'ns2:ReisInformatieProductDVS' or 'ns2:ReisInformatieProductDAS': {\n",
    "      'ns2:DynamischeVertrekStaat' or 'ns2:DynamischeAankomstStaat': {\n",
    "          'ns2:RitStation': <station_info>,\n",
    "          'ns2:Trein' or 'ns2:TreinAankomst': {\n",
    "              'ns2:VertrekTijd' or 'ns2:AankomstTijd': [<planned_and_actual_times>],\n",
    "              'ns2:TreinNummer': <train_number>,\n",
    "              'ns2:TreinSoort': <kind_of_train>,\n",
    "              ...\n",
    "          }\n",
    "           \n",
    "      }\n",
    "    }\n",
    "  }\n",
    "}\n",
    "```\n",
    "\n",
    "We can see also that the train stations have a long name `ns2:LangeNaam`, a medium name `ns2:MiddelNaam`, a short name `ns2:KorteNaam`, a station code `ns2:StationCode` and a kind of nummerical ID `ns2:UICCode`. When giving information about times, tracks, direction,... you will find sometimes the information twice with the status `Gepland` (which means planned, according to the schedule) and `Actueel`(which means the actual measured value). "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### a) Parse - 5/20 \n",
    "\n",
    "We want to compute the time a train stays at a station and get a real-time histogram for a given time window. To begin with, you need to write some parsing functions that will allow you to get information from the data streams. We have prepare one function `parse_train_dep` for the stream `ndovloketnl-departures`, which returns a Key-Value pair."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import json\n",
    "\n",
    "def parse_train_dep(s):\n",
    "    obj = json.loads(s)\n",
    "    tn = (obj.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "             .get('ns2:ReisInformatieProductDVS', {})\n",
    "             .get('ns2:DynamischeVertrekStaat', {})\n",
    "             .get('ns2:Trein', {})\n",
    "             .get(\"ns2:TreinNummer\"))\n",
    "    st = (obj.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "             .get('ns2:ReisInformatieProductDVS', {})\n",
    "             .get('ns2:DynamischeVertrekStaat', {})\n",
    "             .get('ns2:RitStation', {})\n",
    "             .get(\"ns2:UICCode\"))\n",
    "    if tn and st:\n",
    "        return [(\"{}-{}\".format(tn, st), obj)]\n",
    "    else:\n",
    "        return []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('31241-8400169', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:5', u'ns2:ReisInformatieProductDVS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-04-28T12:07:00.000Z', u'ns2:ReisInformatieProductID': u'2104281408062300003', u'ns2:AbonnementId': u'54'}, u'@TimeStamp': u'2021-04-28T12:08:06.847Z', u'@Versie': u'6.2', u'ns2:DynamischeVertrekStaat': {u'ns2:RitId': u'31241', u'ns2:RitStation': {u'ns2:LangeNaam': u'Delden', u'ns2:MiddelNaam': u'Delden', u'ns2:StationCode': u'DDN', u'ns2:Type': u'0', u'ns2:UICCode': u'8400169', u'ns2:KorteNaam': u'Delden'}, u'ns2:RitDatum': u'2021-04-28', u'ns2:Trein': {u'ns2:VertrekRichting': u'B', u'ns2:VertrekTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-04-28T12:07:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-04-28T12:07:00.000Z'}], u'ns2:RangeerBeweging': u'N', u'ns2:PresentatieTreinVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:NietInstappen': u'N', u'ns2:PresentatieVerkorteRoute': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hengelo'}}, u'ns2:TreinStatus': u'5', u'ns2:TreinNummer': u'31241', u'ns2:ExacteVertrekVertraging': u'PT0S', u'ns2:Wijziging': {u'ns2:WijzigingType': u'40'}, u'ns2:TreinVleugel': {u'ns2:PresentatieTreinVleugelVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:TreinVleugelVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:StopStations': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Hengelo Gezondheidspark', u'ns2:MiddelNaam': u'Hengelo Gezondh.', u'ns2:StationCode': u'HGLG', u'ns2:Type': u'0', u'ns2:UICCode': u'8400433', u'ns2:KorteNaam': u'Gezondhprk'}, {u'ns2:LangeNaam': u'Hengelo', u'ns2:MiddelNaam': u'Hengelo', u'ns2:StationCode': u'HGL', u'ns2:Type': u'5', u'ns2:UICCode': u'8400316', u'ns2:KorteNaam': u'Hengelo'}, {u'ns2:LangeNaam': u'Hengelo Oost', u'ns2:MiddelNaam': u'Hengelo O.', u'ns2:StationCode': u'HGLO', u'ns2:Type': u'0', u'ns2:UICCode': u'8400312', u'ns2:KorteNaam': u'Hengelo O'}, {u'ns2:LangeNaam': u'Oldenzaal', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Hengelo Gezondheidspark', u'ns2:MiddelNaam': u'Hengelo Gezondh.', u'ns2:StationCode': u'HGLG', u'ns2:Type': u'0', u'ns2:UICCode': u'8400433', u'ns2:KorteNaam': u'Gezondhprk'}, {u'ns2:LangeNaam': u'Hengelo', u'ns2:MiddelNaam': u'Hengelo', u'ns2:StationCode': u'HGL', u'ns2:Type': u'5', u'ns2:UICCode': u'8400316', u'ns2:KorteNaam': u'Hengelo'}, {u'ns2:LangeNaam': u'Hengelo Oost', u'ns2:MiddelNaam': u'Hengelo O.', u'ns2:StationCode': u'HGLO', u'ns2:Type': u'0', u'ns2:UICCode': u'8400312', u'ns2:KorteNaam': u'Hengelo O'}, {u'ns2:LangeNaam': u'Oldenzaal', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}], u'@InfoStatus': u'Actueel'}], u'ns2:TreinVleugelEindBestemming': [{u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}, {u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}], u'ns2:MaterieelDeelDVS': {u'ns2:PresentatieMaterieelDeelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Oldenzaal'}}, u'ns2:MaterieelLengte': u'4100', u'ns2:MaterieelAanduiding': u'2', u'ns2:MaterieelSoort': u'LINT KE', u'ns2:MaterieelDeelEindBestemming': [{u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}, {u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}]}, u'ns2:PresentatieTreinVleugelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Oldenzaal'}}}, u'ns2:TreinEindBestemming': [{u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}, {u'ns2:LangeNaam': u'Oldenzaal', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Oldenzaal', u'ns2:StationCode': u'ODZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400483', u'ns2:KorteNaam': u'Oldenzaal'}], u'ns2:AchterBlijvenAchtersteTreinDeel': u'N', u'ns2:Reserveren': u'N', u'ns2:AfstandPerronEindKopVertrekTrein': u'0', u'ns2:VerkorteRoute': [{u'ns2:Station': {u'ns2:LangeNaam': u'Hengelo', u'ns2:MiddelNaam': u'Hengelo', u'ns2:StationCode': u'HGL', u'ns2:Type': u'5', u'ns2:UICCode': u'8400316', u'ns2:KorteNaam': u'Hengelo'}, u'@InfoStatus': u'Gepland'}, {u'ns2:Station': {u'ns2:LangeNaam': u'Hengelo', u'ns2:MiddelNaam': u'Hengelo', u'ns2:StationCode': u'HGL', u'ns2:Type': u'5', u'ns2:UICCode': u'8400316', u'ns2:KorteNaam': u'Hengelo'}, u'@InfoStatus': u'Actueel'}], u'ns2:TreinVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:Toeslag': u'N', u'ns2:TreinSoort': {u'@Code': u'ST', u'#text': u'Stoptrein'}, u'ns2:TreinFormule': u'1', u'ns2:PresentatieTreinEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Oldenzaal'}}, u'ns2:SpeciaalKaartje': u'N', u'ns2:Vervoerder': u'Blauwnet', u'ns2:GedempteVertrekVertraging': u'PT0S'}}}, u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4'}})]"
     ]
    }
   ],
   "source": [
    "parse_train_dep(example_departures.value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__TODO - 1/5__ Please check the function `parse_train_dep` above. Explain how we construct the Key and the Value, and why we construct them in this way."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Answer:** In the above function we contruct the key by extracting from the parsed json both the `TreinNummer` and `UICCode` of the current stop station. To extract them we iteratively get the appropriate dictionnary from the parsed JSON, each time returning default empty dictionnary if the key is not found. If the given JSON is missing one of these two attributes we return an empty array, if we managed to extract both `TreinNummer` and `UICCode` we concatenate them (with a ´-´ in between) to form a key and we add the entire parsed json object as value.  \n",
    "This construction allows to keep all informations contained in our parsed JSON and to index it with a key that is suited to compute the time a train stays at a station (as we have both information concatenated as a unique key)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__TODO - 2/5__  Take `parse_train_dep` as an example and write the function `parse_train_arr` for the stream `ndovloketnl-arrivals`. Make sure they have the same output format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def parse_train_arr(s):\n",
    "    obj = json.loads(s)\n",
    "    tn = (obj.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "             .get('ns2:ReisInformatieProductDAS', {})\n",
    "             .get('ns2:DynamischeAankomstStaat', {})\n",
    "             .get('ns2:TreinAankomst', {})\n",
    "             .get(\"ns2:TreinNummer\"))\n",
    "    st = (obj.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "             .get('ns2:ReisInformatieProductDAS', {})\n",
    "             .get('ns2:DynamischeAankomstStaat', {})\n",
    "             .get('ns2:RitStation', {})\n",
    "             .get(\"ns2:UICCode\"))\n",
    "    if tn and st:\n",
    "        return [(\"{}-{}\".format(tn, st), obj)]\n",
    "    else:\n",
    "        return []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('37542-8400696', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1', u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4', u'ns2:ReisInformatieProductDAS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-04-26T11:20:00.000Z', u'ns2:ReisInformatieProductID': u'2104261318312400002', u'ns2:AbonnementId': u'55'}, u'@TimeStamp': u'2021-04-26T11:18:31.081Z', u'@Versie': u'6.1', u'ns2:DynamischeAankomstStaat': {u'ns2:RitId': u'37542', u'ns2:RitStation': {u'ns2:LangeNaam': u'Winschoten', u'ns2:MiddelNaam': u'Winschoten', u'ns2:StationCode': u'WS', u'ns2:Type': u'1', u'ns2:UICCode': u'8400696', u'ns2:KorteNaam': u'Winschoten'}, u'ns2:TreinAankomst': {u'ns2:TreinHerkomst': [{u'ns2:LangeNaam': u'Groningen', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Groningen', u'ns2:StationCode': u'GN', u'ns2:Type': u'5', u'ns2:UICCode': u'8400263', u'ns2:KorteNaam': u'Groningen'}, {u'ns2:LangeNaam': u'Groningen', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Groningen', u'ns2:StationCode': u'GN', u'ns2:Type': u'5', u'ns2:UICCode': u'8400263', u'ns2:KorteNaam': u'Groningen'}], u'ns2:TreinSoort': {u'@Code': u'ST', u'#text': u'Stoptrein'}, u'ns2:WijzigingHerkomst': {u'ns2:WijzigingType': u'40'}, u'ns2:VerkorteRouteHerkomst': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Hoogezand-Sappemeer', u'ns2:MiddelNaam': u'Hoogezand-Sap.', u'ns2:StationCode': u'HGZ', u'ns2:Type': u'1', u'ns2:UICCode': u'8400331', u'ns2:KorteNaam': u'Hoogezand'}, {u'ns2:LangeNaam': u'Zuidbroek', u'ns2:MiddelNaam': u'Zuidbroek', u'ns2:StationCode': u'ZB', u'ns2:Type': u'1', u'ns2:UICCode': u'8400742', u'ns2:KorteNaam': u'Zuidbroek'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Hoogezand-Sappemeer', u'ns2:MiddelNaam': u'Hoogezand-Sap.', u'ns2:StationCode': u'HGZ', u'ns2:Type': u'1', u'ns2:UICCode': u'8400331', u'ns2:KorteNaam': u'Hoogezand'}, {u'ns2:LangeNaam': u'Zuidbroek', u'ns2:MiddelNaam': u'Zuidbroek', u'ns2:StationCode': u'ZB', u'ns2:Type': u'1', u'ns2:UICCode': u'8400742', u'ns2:KorteNaam': u'Zuidbroek'}], u'@InfoStatus': u'Actueel'}], u'ns2:LijnNummer': u'RS6', u'ns2:PresentatieTreinAankomstSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:TreinStatus': u'2', u'ns2:PresentatieVerkorteRouteHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hoogezand-Sap., Zuidbroek'}}, u'ns2:AankomstTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-04-26T11:20:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-04-26T11:20:00.000Z'}], u'ns2:TreinNummer': u'37542', u'ns2:TreinAankomstSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:PresentatieTreinHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Groningen'}}, u'ns2:GedempteAankomstVertraging': u'PT0S', u'ns2:ExacteAankomstVertraging': u'PT0S', u'ns2:Vervoerder': u'Arriva'}, u'ns2:RitDatum': u'2021-04-26'}}}})]"
     ]
    }
   ],
   "source": [
    "parse_train_arr(example_arrivals.value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__TODO - 2/5__ Another parsing function you will need later is `get_actual_time`, which will allow you to extract the **actual** time from the fields of time information, which are `ns2:AankomstTijd` in the arrival stream and `ns2:VertrekTijd` in the departure stream. \n",
    "\n",
    "__Note:__ These two fields may be empty and they may not contain the actual time information. In both cases the function should return `None`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import datetime\n",
    "\n",
    "def get_actual_time(tab): \n",
    "    # As the given lab might not always be ordered in the same way, we iterate over its elements\n",
    "    for field in tab:\n",
    "        # Check if the field corresponds to actual time and has attribute #text\n",
    "        if(field.get('@InfoStatus', None)=='Actueel' and field.get('#text', None) ):\n",
    "            act = field.get('#text')\n",
    "            # Check the date text is non empty, if not returns the corresponding datetime (also take milliseconds into account)\n",
    "            if(len(act)>1): return datetime.datetime.strptime(act[:-1]+\"000\", '%Y-%m-%dT%H:%M:%S.%f')\n",
    "    return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Get the field of time in the departure stream\n",
    "example_dep_json = json.loads(example_departures.value)\n",
    "example_dep_tab = (example_dep_json.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "                                   .get(\"ns2:ReisInformatieProductDVS\", {})\n",
    "                                   .get(\"ns2:DynamischeVertrekStaat\", {})\n",
    "                                   .get(\"ns2:Trein\", {})\n",
    "                                   .get(\"ns2:VertrekTijd\",{}))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "assert get_actual_time(example_dep_tab) == datetime.datetime(2021, 4, 28, 12, 7)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Get the field of time in the arrival stream\n",
    "example_arr_json = json.loads(example_arrivals.value)\n",
    "example_arr_tab = (example_arr_json.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "                                   .get(\"ns2:ReisInformatieProductDAS\", {})\n",
    "                                   .get(\"ns2:DynamischeAankomstStaat\", {})\n",
    "                                   .get(\"ns2:TreinAankomst\", {})\n",
    "                                   .get(\"ns2:AankomstTijd\",{}))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "assert get_actual_time(example_arr_tab) == datetime.datetime(2021, 4, 26, 11, 20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[{u'@InfoStatus': u'Gepland', u'#text': u'2021-04-26T11:20:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-04-26T11:20:00.000Z'}]"
     ]
    }
   ],
   "source": [
    "example_arr_json.get('ns1:PutReisInformatieBoodschapIn').get(\"ns2:ReisInformatieProductDAS\", {}).get(\"ns2:DynamischeAankomstStaat\", {}).get(\"ns2:TreinAankomst\", {}).get(\"ns2:AankomstTijd\",{})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### b) Transform - 5/20\n",
    "\n",
    "Create two Spark streams from the arrivals and departures where the records are in the form (Key, Value) using `parse_train_dep` and  `parse_train_arr`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "checkpoint created at hdfs:///user/cadei/checkpoint/"
     ]
    }
   ],
   "source": [
    "# Redefines the checkpoint folder in case the intro cells were not all run\n",
    "checkpoint = 'hdfs:///user/{}/checkpoint/'.format(username)\n",
    "print('checkpoint created at hdfs:///user/{}/checkpoint/'.format(username))\n",
    "\n",
    "# Init new StreamingContext, sets checkpoints & group\n",
    "ssc = StreamingContext(sc, 10)\n",
    "ssc.checkpoint(checkpoint)\n",
    "group_id = 'ns-{0}'.format(username)\n",
    "\n",
    "# Init input streams\n",
    "raw_arrival_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-arrivals': 1})\n",
    "raw_departure_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-departures': 1})\n",
    "\n",
    "# Apply our parsing on the streams \n",
    "arrival_stream = raw_arrival_stream.flatMap(lambda x: parse_train_arr(x[1]))\n",
    "departure_stream = raw_departure_stream.flatMap(lambda x: parse_train_dep(x[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# TODO: Can be removed from final submission\n",
    "arrival_stream.pprint(num=2) # print the first 2 messages\n",
    "departure_stream.pprint(num=2) # print the first 2 messages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:39:50\n",
      "-------------------------------------------\n",
      "('3074-8400309', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1', u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4', u'ns2:ReisInformatieProductDAS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-05-11T20:40:00.000Z', u'ns2:ReisInformatieProductID': u'2105112239412400004', u'ns2:AbonnementId': u'55'}, u'@TimeStamp': u'2021-05-11T20:39:41.375Z', u'@Versie': u'6.1', u'ns2:DynamischeAankomstStaat': {u'ns2:RitId': u'3074', u'ns2:RitStation': {u'ns2:LangeNaam': u'Heiloo', u'ns2:MiddelNaam': u'Heiloo', u'ns2:StationCode': u'HLO', u'ns2:Type': u'0', u'ns2:UICCode': u'8400309', u'ns2:KorteNaam': u'Heiloo'}, u'ns2:TreinAankomst': {u'ns2:TreinHerkomst': [{u'ns2:LangeNaam': u'Nijmegen', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Nijmegen', u'ns2:StationCode': u'NM', u'ns2:Type': u'5', u'ns2:UICCode': u'8400470', u'ns2:KorteNaam': u'Nijmegen'}, {u'ns2:LangeNaam': u'Nijmegen', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Nijmegen', u'ns2:StationCode': u'NM', u'ns2:Type': u'5', u'ns2:UICCode': u'8400470', u'ns2:KorteNaam': u'Nijmegen'}], u'ns2:TreinSoort': {u'@Code': u'IC', u'#text': u'Intercity'}, u'ns2:WijzigingHerkomst': {u'ns2:WijzigingType': u'40'}, u'ns2:VerkorteRouteHerkomst': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Utrecht Centraal', u'ns2:MiddelNaam': u'Utrecht C.', u'ns2:StationCode': u'UT', u'ns2:Type': u'6', u'ns2:UICCode': u'8400621', u'ns2:KorteNaam': u'Utrecht C'}, {u'ns2:LangeNaam': u'Amsterdam Centraal', u'ns2:MiddelNaam': u'Amsterdam C.', u'ns2:StationCode': u'ASD', u'ns2:Type': u'6', u'ns2:UICCode': u'8400058', u'ns2:KorteNaam': u'Amsterdm C'}, {u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Castricum', u'ns2:MiddelNaam': u'Castricum', u'ns2:StationCode': u'CAS', u'ns2:Type': u'4', u'ns2:UICCode': u'8400151', u'ns2:KorteNaam': u'Castricum'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Utrecht Centraal', u'ns2:MiddelNaam': u'Utrecht C.', u'ns2:StationCode': u'UT', u'ns2:Type': u'6', u'ns2:UICCode': u'8400621', u'ns2:KorteNaam': u'Utrecht C'}, {u'ns2:LangeNaam': u'Amsterdam Centraal', u'ns2:MiddelNaam': u'Amsterdam C.', u'ns2:StationCode': u'ASD', u'ns2:Type': u'6', u'ns2:UICCode': u'8400058', u'ns2:KorteNaam': u'Amsterdm C'}, {u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Castricum', u'ns2:MiddelNaam': u'Castricum', u'ns2:StationCode': u'CAS', u'ns2:Type': u'4', u'ns2:UICCode': u'8400151', u'ns2:KorteNaam': u'Castricum'}], u'@InfoStatus': u'Actueel'}], u'ns2:PresentatieTreinAankomstSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:TreinStatus': u'2', u'ns2:PresentatieVerkorteRouteHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Utrecht C., Amsterdam C., Zaandam, Castricum'}}, u'ns2:AankomstTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-05-11T20:40:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-05-11T20:40:00.000Z'}], u'ns2:TreinNummer': u'3074', u'ns2:TreinAankomstSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:PresentatieTreinHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Nijmegen'}}, u'ns2:GedempteAankomstVertraging': u'PT0S', u'ns2:ExacteAankomstVertraging': u'PT0S', u'ns2:Vervoerder': u'NS'}, u'ns2:RitDatum': u'2021-05-11'}}}})\n",
      "('3387-8400499', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:dynamischeaankomststaat:1', u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4', u'ns2:ReisInformatieProductDAS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-05-11T20:41:00.000Z', u'ns2:ReisInformatieProductID': u'2105112239452500004', u'ns2:AbonnementId': u'55'}, u'@TimeStamp': u'2021-05-11T20:39:45.282Z', u'@Versie': u'6.1', u'ns2:DynamischeAankomstStaat': {u'ns2:RitId': u'3387', u'ns2:RitStation': {u'ns2:LangeNaam': u'Purmerend Weidevenne', u'ns2:MiddelNaam': u'Weidevenne', u'ns2:StationCode': u'PMW', u'ns2:Type': u'0', u'ns2:UICCode': u'8400499', u'ns2:KorteNaam': u'Weidevenne'}, u'ns2:TreinAankomst': {u'ns2:TreinHerkomst': [{u'ns2:LangeNaam': u'Hoorn Kersenboogerd', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Kersenboogerd', u'ns2:StationCode': u'HNK', u'ns2:Type': u'0', u'ns2:UICCode': u'8400336', u'ns2:KorteNaam': u'Hoorn Kers'}, {u'ns2:LangeNaam': u'Hoorn Kersenboogerd', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Kersenboogerd', u'ns2:StationCode': u'HNK', u'ns2:Type': u'0', u'ns2:UICCode': u'8400336', u'ns2:KorteNaam': u'Hoorn Kers'}], u'ns2:TreinSoort': {u'@Code': u'SPR', u'#text': u'Sprinter'}, u'ns2:WijzigingHerkomst': {u'ns2:WijzigingType': u'40'}, u'ns2:VerkorteRouteHerkomst': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Hoorn', u'ns2:MiddelNaam': u'Hoorn', u'ns2:StationCode': u'HN', u'ns2:Type': u'5', u'ns2:UICCode': u'8400337', u'ns2:KorteNaam': u'Hoorn'}, {u'ns2:LangeNaam': u'Purmerend', u'ns2:MiddelNaam': u'Purmerend', u'ns2:StationCode': u'PMR', u'ns2:Type': u'1', u'ns2:UICCode': u'8400508', u'ns2:KorteNaam': u'Purmerend'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Hoorn', u'ns2:MiddelNaam': u'Hoorn', u'ns2:StationCode': u'HN', u'ns2:Type': u'5', u'ns2:UICCode': u'8400337', u'ns2:KorteNaam': u'Hoorn'}, {u'ns2:LangeNaam': u'Purmerend', u'ns2:MiddelNaam': u'Purmerend', u'ns2:StationCode': u'PMR', u'ns2:Type': u'1', u'ns2:UICCode': u'8400508', u'ns2:KorteNaam': u'Purmerend'}], u'@InfoStatus': u'Actueel'}], u'ns2:PresentatieTreinAankomstSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:TreinStatus': u'2', u'ns2:PresentatieVerkorteRouteHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hoorn, Purmerend'}}, u'ns2:AankomstTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-05-11T20:41:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-05-11T20:41:00.000Z'}], u'ns2:TreinNummer': u'3387', u'ns2:TreinAankomstSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:PresentatieTreinHerkomst': {u'ns2:Uitingen': {u'ns2:Uiting': u'Kersenboogerd'}}, u'ns2:GedempteAankomstVertraging': u'PT0S', u'ns2:ExacteAankomstVertraging': u'PT0S', u'ns2:Vervoerder': u'NS'}, u'ns2:RitDatum': u'2021-05-11'}}}})\n",
      "...\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:39:50\n",
      "-------------------------------------------\n",
      "('31371-8400055', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:5', u'ns2:ReisInformatieProductDVS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-05-11T20:39:00.000Z', u'ns2:ReisInformatieProductID': u'2105112239452500002', u'ns2:AbonnementId': u'54'}, u'@TimeStamp': u'2021-05-11T20:39:45.238Z', u'@Versie': u'6.2', u'ns2:DynamischeVertrekStaat': {u'ns2:RitId': u'31371', u'ns2:RitStation': {u'ns2:LangeNaam': u'Amersfoort Centraal', u'ns2:MiddelNaam': u'Amersfoort C.', u'ns2:StationCode': u'AMF', u'ns2:Type': u'5', u'ns2:UICCode': u'8400055', u'ns2:KorteNaam': u'Amersfrt C'}, u'ns2:RitDatum': u'2021-05-11', u'ns2:Trein': {u'ns2:TreinVleugel': {u'ns2:PresentatieTreinVleugelVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'4b'}}, u'ns2:TreinVleugelVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorFase': u'b', u'ns2:SpoorNummer': u'4'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorFase': u'b', u'ns2:SpoorNummer': u'4'}], u'ns2:StopStations': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Hoevelaken', u'ns2:MiddelNaam': u'Hoevelaken', u'ns2:StationCode': u'HVL', u'ns2:Type': u'0', u'ns2:UICCode': u'8400490', u'ns2:KorteNaam': u'Hoevelaken'}, {u'ns2:LangeNaam': u'Barneveld Noord', u'ns2:MiddelNaam': u'Barneveld N.', u'ns2:StationCode': u'BNN', u'ns2:Type': u'0', u'ns2:UICCode': u'8400092', u'ns2:KorteNaam': u'Barnevld N'}, {u'ns2:LangeNaam': u'Barneveld Centrum', u'ns2:MiddelNaam': u'Barneveld C.', u'ns2:StationCode': u'BNC', u'ns2:Type': u'0', u'ns2:UICCode': u'8400091', u'ns2:KorteNaam': u'Barnevld C'}, {u'ns2:LangeNaam': u'Barneveld Zuid', u'ns2:MiddelNaam': u'Barneveld Z.', u'ns2:StationCode': u'BNZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400115', u'ns2:KorteNaam': u'Barnevld Z'}, {u'ns2:LangeNaam': u'Lunteren', u'ns2:MiddelNaam': u'Lunteren', u'ns2:StationCode': u'LTN', u'ns2:Type': u'0', u'ns2:UICCode': u'8400405', u'ns2:KorteNaam': u'Lunteren'}, {u'ns2:LangeNaam': u'Ede Centrum', u'ns2:MiddelNaam': u'Ede C.', u'ns2:StationCode': u'EDC', u'ns2:Type': u'0', u'ns2:UICCode': u'8400201', u'ns2:KorteNaam': u'Ede C'}, {u'ns2:LangeNaam': u'Ede-Wageningen', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Hoevelaken', u'ns2:MiddelNaam': u'Hoevelaken', u'ns2:StationCode': u'HVL', u'ns2:Type': u'0', u'ns2:UICCode': u'8400490', u'ns2:KorteNaam': u'Hoevelaken'}, {u'ns2:LangeNaam': u'Barneveld Noord', u'ns2:MiddelNaam': u'Barneveld N.', u'ns2:StationCode': u'BNN', u'ns2:Type': u'0', u'ns2:UICCode': u'8400092', u'ns2:KorteNaam': u'Barnevld N'}, {u'ns2:LangeNaam': u'Barneveld Centrum', u'ns2:MiddelNaam': u'Barneveld C.', u'ns2:StationCode': u'BNC', u'ns2:Type': u'0', u'ns2:UICCode': u'8400091', u'ns2:KorteNaam': u'Barnevld C'}, {u'ns2:LangeNaam': u'Barneveld Zuid', u'ns2:MiddelNaam': u'Barneveld Z.', u'ns2:StationCode': u'BNZ', u'ns2:Type': u'0', u'ns2:UICCode': u'8400115', u'ns2:KorteNaam': u'Barnevld Z'}, {u'ns2:LangeNaam': u'Lunteren', u'ns2:MiddelNaam': u'Lunteren', u'ns2:StationCode': u'LTN', u'ns2:Type': u'0', u'ns2:UICCode': u'8400405', u'ns2:KorteNaam': u'Lunteren'}, {u'ns2:LangeNaam': u'Ede Centrum', u'ns2:MiddelNaam': u'Ede C.', u'ns2:StationCode': u'EDC', u'ns2:Type': u'0', u'ns2:UICCode': u'8400201', u'ns2:KorteNaam': u'Ede C'}, {u'ns2:LangeNaam': u'Ede-Wageningen', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}], u'@InfoStatus': u'Actueel'}], u'ns2:TreinVleugelEindBestemming': [{u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}, {u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}], u'ns2:MaterieelDeelDVS': {u'ns2:PresentatieMaterieelDeelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Ede-Wageningen'}}, u'ns2:MaterieelLengte': u'6300', u'ns2:MaterieelAanduiding': u'3', u'ns2:MaterieelSoort': u'FLIRT CO', u'ns2:MaterieelDeelEindBestemming': [{u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}, {u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}]}, u'ns2:PresentatieTreinVleugelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Ede-Wageningen'}}}, u'ns2:VertrekTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-05-11T20:39:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-05-11T20:39:00.000Z'}], u'ns2:RangeerBeweging': u'N', u'ns2:PresentatieTreinVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'4b'}}, u'ns2:NietInstappen': u'N', u'ns2:TreinStatus': u'5', u'ns2:TreinNummer': u'31371', u'ns2:ExacteVertrekVertraging': u'PT0S', u'ns2:Wijziging': {u'ns2:WijzigingType': u'40'}, u'ns2:VertrekRichting': u'B', u'ns2:TreinEindBestemming': [{u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}, {u'ns2:LangeNaam': u'Ede-Wageningen', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Ede-Wageningen', u'ns2:StationCode': u'ED', u'ns2:Type': u'5', u'ns2:UICCode': u'8400200', u'ns2:KorteNaam': u'Ede-Wag'}], u'ns2:AchterBlijvenAchtersteTreinDeel': u'N', u'ns2:Reserveren': u'N', u'ns2:AfstandPerronEindKopVertrekTrein': u'0', u'ns2:GedempteVertrekVertraging': u'PT0S', u'ns2:TreinVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorFase': u'b', u'ns2:SpoorNummer': u'4'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorFase': u'b', u'ns2:SpoorNummer': u'4'}], u'ns2:Toeslag': u'N', u'ns2:TreinSoort': {u'@Code': u'ST', u'#text': u'Stoptrein'}, u'ns2:TreinFormule': u'1', u'ns2:PresentatieTreinEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Ede-Wageningen'}}, u'ns2:SpeciaalKaartje': u'N', u'ns2:Vervoerder': u'Valleilijn'}}}, u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4'}})\n",
      "('3387-8400499', {u'ns1:PutReisInformatieBoodschapIn': {u'@xmlns:ns1': u'urn:ndov:cdm:trein:reisinformatie:messages:5', u'ns2:ReisInformatieProductDVS': {u'ns2:RIPAdministratie': {u'ns2:ReisInformatieTijdstip': u'2021-05-11T20:41:00.000Z', u'ns2:ReisInformatieProductID': u'2105112239452500003', u'ns2:AbonnementId': u'54'}, u'@TimeStamp': u'2021-05-11T20:39:45.274Z', u'@Versie': u'6.2', u'ns2:DynamischeVertrekStaat': {u'ns2:RitId': u'3387', u'ns2:RitStation': {u'ns2:LangeNaam': u'Purmerend Weidevenne', u'ns2:MiddelNaam': u'Weidevenne', u'ns2:StationCode': u'PMW', u'ns2:Type': u'0', u'ns2:UICCode': u'8400499', u'ns2:KorteNaam': u'Weidevenne'}, u'ns2:RitDatum': u'2021-05-11', u'ns2:Trein': {u'ns2:VertrekRichting': u'B', u'ns2:VertrekTijd': [{u'@InfoStatus': u'Gepland', u'#text': u'2021-05-11T20:41:00.000Z'}, {u'@InfoStatus': u'Actueel', u'#text': u'2021-05-11T20:41:00.000Z'}], u'ns2:RangeerBeweging': u'N', u'ns2:PresentatieTreinVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:NietInstappen': u'N', u'ns2:PresentatieVerkorteRoute': {u'ns2:Uitingen': {u'ns2:Uiting': u'Zaandam, Sloterdijk, Schiphol Airport'}}, u'ns2:TreinStatus': u'2', u'ns2:TreinNummer': u'3387', u'ns2:ExacteVertrekVertraging': u'PT0S', u'ns2:Wijziging': {u'ns2:WijzigingType': u'40'}, u'ns2:TreinVleugel': {u'ns2:PresentatieTreinVleugelVertrekSpoor': {u'ns2:Uitingen': {u'ns2:Uiting': u'1'}}, u'ns2:TreinVleugelVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:StopStations': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Zaandam Kogerveld', u'ns2:MiddelNaam': u'Kogerveld', u'ns2:StationCode': u'ZDK', u'ns2:Type': u'0', u'ns2:UICCode': u'8400728', u'ns2:KorteNaam': u'Kogerveld'}, {u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Amsterdam Sloterdijk', u'ns2:MiddelNaam': u'Sloterdijk', u'ns2:StationCode': u'ASS', u'ns2:Type': u'5', u'ns2:UICCode': u'8400059', u'ns2:KorteNaam': u'Sloterdijk'}, {u'ns2:LangeNaam': u'Amsterdam Lelylaan', u'ns2:MiddelNaam': u'Lelylaan', u'ns2:StationCode': u'ASDL', u'ns2:Type': u'0', u'ns2:UICCode': u'8400079', u'ns2:KorteNaam': u'Lelylaan'}, {u'ns2:LangeNaam': u'Schiphol Airport', u'ns2:MiddelNaam': u'Schiphol Airport', u'ns2:StationCode': u'SHL', u'ns2:Type': u'6', u'ns2:UICCode': u'8400561', u'ns2:KorteNaam': u'Schiphol'}, {u'ns2:LangeNaam': u'Hoofddorp', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Zaandam Kogerveld', u'ns2:MiddelNaam': u'Kogerveld', u'ns2:StationCode': u'ZDK', u'ns2:Type': u'0', u'ns2:UICCode': u'8400728', u'ns2:KorteNaam': u'Kogerveld'}, {u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Amsterdam Sloterdijk', u'ns2:MiddelNaam': u'Sloterdijk', u'ns2:StationCode': u'ASS', u'ns2:Type': u'5', u'ns2:UICCode': u'8400059', u'ns2:KorteNaam': u'Sloterdijk'}, {u'ns2:LangeNaam': u'Amsterdam Lelylaan', u'ns2:MiddelNaam': u'Lelylaan', u'ns2:StationCode': u'ASDL', u'ns2:Type': u'0', u'ns2:UICCode': u'8400079', u'ns2:KorteNaam': u'Lelylaan'}, {u'ns2:LangeNaam': u'Schiphol Airport', u'ns2:MiddelNaam': u'Schiphol Airport', u'ns2:StationCode': u'SHL', u'ns2:Type': u'6', u'ns2:UICCode': u'8400561', u'ns2:KorteNaam': u'Schiphol'}, {u'ns2:LangeNaam': u'Hoofddorp', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}], u'@InfoStatus': u'Actueel'}], u'ns2:TreinVleugelEindBestemming': [{u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}, {u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}], u'ns2:MaterieelDeelDVS': {u'ns2:MaterieelLengte': u'7576', u'ns2:MaterieelNummer': u'000000-02709-0', u'ns2:PresentatieMaterieelDeelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hoofddorp'}}, u'ns2:MaterieelDeelEindBestemming': [{u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}, {u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}], u'ns2:MaterieelAanduiding': u'4', u'ns2:MaterieelSoort': u'SNG'}, u'ns2:PresentatieTreinVleugelEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hoofddorp'}}}, u'ns2:TreinEindBestemming': [{u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Gepland', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}, {u'ns2:LangeNaam': u'Hoofddorp', u'@InfoStatus': u'Actueel', u'ns2:MiddelNaam': u'Hoofddorp', u'ns2:StationCode': u'HFD', u'ns2:Type': u'0', u'ns2:UICCode': u'8400332', u'ns2:KorteNaam': u'Hoofddorp'}], u'ns2:AchterBlijvenAchtersteTreinDeel': u'N', u'ns2:Reserveren': u'N', u'ns2:TreinFormule': u'1', u'ns2:VerkorteRoute': [{u'ns2:Station': [{u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Amsterdam Sloterdijk', u'ns2:MiddelNaam': u'Sloterdijk', u'ns2:StationCode': u'ASS', u'ns2:Type': u'5', u'ns2:UICCode': u'8400059', u'ns2:KorteNaam': u'Sloterdijk'}, {u'ns2:LangeNaam': u'Schiphol Airport', u'ns2:MiddelNaam': u'Schiphol Airport', u'ns2:StationCode': u'SHL', u'ns2:Type': u'6', u'ns2:UICCode': u'8400561', u'ns2:KorteNaam': u'Schiphol'}], u'@InfoStatus': u'Gepland'}, {u'ns2:Station': [{u'ns2:LangeNaam': u'Zaandam', u'ns2:MiddelNaam': u'Zaandam', u'ns2:StationCode': u'ZD', u'ns2:Type': u'5', u'ns2:UICCode': u'8400731', u'ns2:KorteNaam': u'Zaandam'}, {u'ns2:LangeNaam': u'Amsterdam Sloterdijk', u'ns2:MiddelNaam': u'Sloterdijk', u'ns2:StationCode': u'ASS', u'ns2:Type': u'5', u'ns2:UICCode': u'8400059', u'ns2:KorteNaam': u'Sloterdijk'}, {u'ns2:LangeNaam': u'Schiphol Airport', u'ns2:MiddelNaam': u'Schiphol Airport', u'ns2:StationCode': u'SHL', u'ns2:Type': u'6', u'ns2:UICCode': u'8400561', u'ns2:KorteNaam': u'Schiphol'}], u'@InfoStatus': u'Actueel'}], u'ns2:TreinVertrekSpoor': [{u'@InfoStatus': u'Gepland', u'ns2:SpoorNummer': u'1'}, {u'@InfoStatus': u'Actueel', u'ns2:SpoorNummer': u'1'}], u'ns2:Toeslag': u'N', u'ns2:TreinSoort': {u'@Code': u'SPR', u'#text': u'Sprinter'}, u'ns2:AfstandPerronEindKopVertrekTrein': u'0', u'ns2:PresentatieTreinEindBestemming': {u'ns2:Uitingen': {u'ns2:Uiting': u'Hoofddorp'}}, u'ns2:SpeciaalKaartje': u'N', u'ns2:Vervoerder': u'NS', u'ns2:GedempteVertrekVertraging': u'PT0S'}}}, u'@xmlns:ns2': u'urn:ndov:cdm:trein:reisinformatie:data:4'}})\n",
      "..."
     ]
    }
   ],
   "source": [
    "ssc.start()\n",
    "ssc.awaitTermination(timeout=10)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "ssc.stop(stopSparkContext=False, stopGraceFully=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### c) Window and Join - 5/20\n",
    "\n",
    "Every 20 seconds, we want to have a list of trains that had departed from any train station after staying for 5 minutes or less at the station. Apply a window of 20s sliding interval on arrival and departure streams. Join two streams such that trains staying for 5 minutes or less (± 20 seconds error due to sliding interval) at any station are caught in the RDD window of the joined stream (you can ignore late messages). \n",
    "\n",
    "__Note:__\n",
    "- Check [here](https://spark.apache.org/docs/2.3.2/streaming-programming-guide.html#window-operations) for windowed computations in Spark Streaming.\n",
    "- Use the methods [reduceByKeyAndWindow](https://spark.apache.org/docs/2.3.2/api/python/pyspark.streaming.html?highlight=reducebykey#pyspark.streaming.DStream.reduceByKeyAndWindow) and [join](https://spark.apache.org/docs/2.3.2/api/python/pyspark.streaming.html?highlight=reducebykey#pyspark.streaming.DStream.join) on DStream objects.\n",
    "- Both windows should have `slideDuration` of 20s\n",
    "- You have to pick the sizes of windows `windowDuration` carefully. The sizes can be different: \n",
    "    - The trains staying for 5 minutes or less (± 20 seconds error due to sliding interval) must be in the joined stream.\n",
    "    - A same stay (i.e. one train at one station) is caught in the joined stream once and only once."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "\n",
    "def to_time_arrival(x):   \n",
    "    example_arr_tab = (x.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "                                   .get(\"ns2:ReisInformatieProductDAS\", {})\n",
    "                                   .get(\"ns2:DynamischeAankomstStaat\", {})\n",
    "                                   .get(\"ns2:TreinAankomst\", {})\n",
    "                                   .get(\"ns2:AankomstTijd\",{}))\n",
    "    return get_actual_time(example_arr_tab)\n",
    "\n",
    "def to_time_departure(x):   \n",
    "    example_arr_tab = (x.get('ns1:PutReisInformatieBoodschapIn', {})\n",
    "                                   .get(\"ns2:ReisInformatieProductDVS\", {})\n",
    "                                   .get(\"ns2:DynamischeVertrekStaat\", {})\n",
    "                                   .get(\"ns2:Trein\", {})\n",
    "                                   .get(\"ns2:VertrekTijd\",{}))\n",
    "    return get_actual_time(example_arr_tab)\n",
    "\n",
    "def time(x):\n",
    "    key, values = x\n",
    "    departure, arrival = values\n",
    "    diff = departure['time']-arrival['time']\n",
    "    \n",
    "    if (diff.total_seconds()/60) <= 5:\n",
    "        return [(key,int(diff.total_seconds()/60))]\n",
    "    else:\n",
    "        return []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Init new StreamingContext, sets checkpoints & group\n",
    "ssc = StreamingContext(sc, 20)\n",
    "checkpoint = 'hdfs:///user/{}/checkpoint/'.format(username)\n",
    "ssc.checkpoint(checkpoint)\n",
    "group_id = 'ns-{0}'.format(username)\n",
    "\n",
    "# Init input streams\n",
    "raw_arrival_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-arrivals': 1})\n",
    "raw_departure_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-departures': 1})\n",
    "\n",
    "# Apply our parsing on the streams \n",
    "arrival_stream = raw_arrival_stream.flatMap(lambda x: parse_train_arr(x[1]))\n",
    "departure_stream = raw_departure_stream.flatMap(lambda x: parse_train_dep(x[1]))\n",
    "\n",
    "arrival_stream = arrival_stream.map(lambda x: (x[0],{'time' : to_time_arrival(x[1])}))\n",
    "departure_stream = departure_stream.map(lambda x: (x[0],{'time' : to_time_departure(x[1])}))\n",
    "\n",
    "arrival_window = arrival_stream.reduceByKeyAndWindow(lambda x,y : x, lambda x, y: x, 5*60+20, 20)\n",
    "departure_window = departure_stream.reduceByKeyAndWindow(lambda x,y : x, lambda x, y: x, 20, 20)\n",
    "\n",
    "joined = departure_window.join(arrival_window).flatMap(time)\n",
    "\n",
    "joined.pprint(num=30)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:40:00\n",
      "-------------------------------------------"
     ]
    }
   ],
   "source": [
    "ssc.start()\n",
    "ssc.awaitTermination(timeout=20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "ssc.stop(stopSparkContext=False, stopGraceFully=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Answer\n",
    "We want to capture trains that stayed in the station for 5 minutes or less, and only count them once. We set the window duration of arrivals at 5 minutes and 20 seconds, the window duration of departures at 20 seconds and the slide duration at 20 seconds. Doing that we effectively have a tumbling window that can capture a train only once. Adding 20 seconds to the arrivals assure that we do not make error due to the sliding interval, and we need to check in the _time()_ function if the stay was indeed less than 5 minutes. With a tumbling window we assure that we capture each train only once. \n",
    "\n",
    "\n",
    "![title](diagram.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### d) Histogram - 5/20\n",
    "\n",
    "On the joined stream, compute the length of each stay (you can round to the minute) and produce a stream of histograms. You don't need to plot them, a value/count array is enough, like:\n",
    "```\n",
    "-------------------------------------------\n",
    "Time: 2018-05-17 11:10:00\n",
    "-------------------------------------------\n",
    "(0.0, 110)\n",
    "(4.0, 3)\n",
    "(8.0, 2) # introduced by late messages\n",
    "\n",
    "-------------------------------------------\n",
    "Time: 2018-05-17 11:10:20\n",
    "-------------------------------------------\n",
    "(0.0, 46)\n",
    "(4.0, 2)\n",
    "(1.0, 5)\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Init new StreamingContext, sets checkpoints & group\n",
    "ssc = StreamingContext(sc, 20)\n",
    "checkpoint = 'hdfs:///user/{}/checkpoint/'.format(username)\n",
    "ssc.checkpoint(checkpoint)\n",
    "group_id = 'ns-{0}'.format(username)\n",
    "\n",
    "# Init input streams\n",
    "raw_arrival_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-arrivals': 1})\n",
    "raw_departure_stream = KafkaUtils.createStream(ssc, ZOOKEEPER_QUORUM, group_id, {'ndovloketnl-departures': 1})\n",
    "\n",
    "# Apply our parsing on the streams \n",
    "arrival_stream = raw_arrival_stream.flatMap(lambda x: parse_train_arr(x[1]))\n",
    "departure_stream = raw_departure_stream.flatMap(lambda x: parse_train_dep(x[1]))\n",
    "\n",
    "arrival_stream = arrival_stream.map(lambda x: (x[0],{'time' : to_time_arrival(x[1])}))\n",
    "departure_stream = departure_stream.map(lambda x: (x[0],{'time' : to_time_departure(x[1])}))\n",
    "\n",
    "arrival_window = arrival_stream.reduceByKeyAndWindow(lambda x,y : x, lambda x, y: x, 5*60, 20)\n",
    "departure_window = departure_stream.reduceByKeyAndWindow(lambda x,y : x, lambda x, y: x, 20, 20)\n",
    "\n",
    "joined = departure_window.join(arrival_window).flatMap(time)\n",
    "\n",
    "histograms_stream = joined.map(lambda x: x[1]).countByValueAndWindow(20,20)\n",
    "histograms_stream.pprint(num=6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:40:40\n",
      "-------------------------------------------\n",
      "(0, 22)\n",
      "(1, 8)\n",
      "(2, 6)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:41:00\n",
      "-------------------------------------------\n",
      "(0, 11)\n",
      "(1, 5)\n",
      "(2, 1)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2021-05-11 22:41:20\n",
      "-------------------------------------------\n",
      "(0, 35)\n",
      "(1, 6)\n",
      "(2, 5)\n",
      "(3, 1)"
     ]
    }
   ],
   "source": [
    "ssc.start() \n",
    "ssc.awaitTermination(timeout=60)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "ssc.stop(stopSparkContext=False, stopGraceFully=False)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}