Spark Streaming application.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark --py-files ./v0.7.0.zip --packages anguenot/pyspark-cassandra:0.9.0,org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.3 --conf spark.cassandra.connection.host=127.0.0.1\n",
    "import findspark\n",
    "findspark.init()\n",
    "\n",
    "import os\n",
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pyspark.sql import Row, SparkSession\n",
    "from pyspark.streaming.kafka import KafkaUtils\n",
    "import pyspark\n",
    "import json\n",
    "import sys\n",
    "import operator\n",
    "import datetime\n",
    "from dateutil import parser\n",
    "from pyspark_cassandra import streaming\n",
    "\n",
    "sc.setLogLevel(\"WARN\")\n",
    "ssc = StreamingContext(sc, 60)\n",
    "\n",
    "# defining the checkpoint directory\n",
    "ssc.checkpoint(\"/tmp\")\n",
    "\n",
    "def getSparkSessionInstance(sparkConf):\n",
    "    if ('sparkSessionSingletonInstance' not in globals()):\n",
    "        globals()['sparkSessionSingletonInstance'] = SparkSession\\\n",
    "            .builder\\\n",
    "            .config(conf=sparkConf)\\\n",
    "            .getOrCreate()\n",
    "    return globals()['sparkSessionSingletonInstance']\n",
    "\n",
    "kafkaStream = KafkaUtils.createStream(ssc, '172.30.7.114:2181,172.30.6.85:2181,172.30.9.172:2181', 'spark-streaming', {'test':1})\n",
    "parsed = kafkaStream.map(lambda v: json.loads(v[1]))\n",
    "\n",
    "# define the update function\n",
    "def update_Q1_1(newValues, runningCount):\n",
    "    if runningCount is None:\n",
    "       runningCount = 0\n",
    "    return sum(newValues, runningCount)\n",
    "\n",
    "# define the update function\n",
    "def update_Q1_2(newValues, runningState):\n",
    "    if runningState is None:\n",
    "       runningState = [0,0]\n",
    "    runningState = list(runningState)\n",
    "    for i in newValues:\n",
    "        runningState[0] += i[0]\n",
    "        runningState[1] += i[1]\n",
    "    return tuple(runningState)\n",
    "\n",
    "def process1_1(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "\n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(id=w[0], count=w[1], airport=w[2]))\n",
    "        df = spark.createDataFrame(rowRdd)\n",
    "        df.write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q1_1\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()\n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass\n",
    "\n",
    "def process1_2(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "        \n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(id=w[0], arrival_delay=w[1], carrier=w[2]))\n",
    "        df = spark.createDataFrame(rowRdd)\n",
    "        df.write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q1_2\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()        \n",
    "\n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass\n",
    "\n",
    "\n",
    "def update_Q2(newValues, runningState):\n",
    "    if runningState is None:\n",
    "       runningState = [0,0]\n",
    "    runningState = list(runningState)\n",
    "    for i in newValues:\n",
    "        runningState[0] += i[0]\n",
    "        runningState[1] += i[1]\n",
    "    return tuple(runningState)\n",
    "\n",
    "def update_Q3(newValues, runningState):\n",
    "    if runningState is None:\n",
    "       runningState = []\n",
    "    if newValues is None:\n",
    "        return None\n",
    "    return runningState + newValues\n",
    "\n",
    "def process2_1(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "        \n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(airport=w[0], departure_delay=w[1][1], carrier=w[1][0]))\n",
    "        df = spark.createDataFrame(rowRdd)\n",
    "        df.write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q2_1\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()         \n",
    "        \n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass\n",
    "\n",
    "def process2_2(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "        \n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(airport=w[0], departure_delay=w[1][1], destination_airport=w[1][0]))\n",
    "        df = spark.createDataFrame(rowRdd)\n",
    "        df.write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q2_2\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()         \n",
    "\n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass\n",
    "\n",
    "def process2_3(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "        \n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(airport=w[0], destination_airport=w[1], arrival_delay=w[3], carrier=w[2]))\n",
    "        df = spark.createDataFrame(rowRdd)\n",
    "        df.write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q2_3\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()         \n",
    "\n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass   \n",
    "\n",
    "# ======================================================================================\n",
    "# Question 1.1 - Rank the top 10 most popular airports by numbers of flights to/from the airport.\n",
    "# ======================================================================================\n",
    "# CREATE TABLE test.q1_1 (id int, count int, airport text, PRIMARY KEY(id, count, airport));\n",
    "# SELECT * FROM test.q1_1 where id = 1 ORDER BY count DESC LIMIT 10;\n",
    "def Q1_1(parsed):\n",
    "    parsed.flatMap(lambda v: [v[\"Origin\"], v[\"Dest\"]]) \\\n",
    "       .map(lambda airport: (airport, 1)) \\\n",
    "       .reduceByKey(lambda a, b: a + b) \\\n",
    "       .updateStateByKey(update_Q1_1) \\\n",
    "       .map(lambda v: (1, v[1], v[0])) \\\n",
    "       .foreachRDD(process1_1)\n",
    "\n",
    "# ======================================================================================\n",
    "# Question 1.2 - Rank the top 10 airlines by on-time arrival performance.\n",
    "# ======================================================================================\n",
    "# CREATE TABLE test.q1_2 (id int, arrival_delay float, carrier text, PRIMARY KEY(id, arrival_delay, carrier));\n",
    "# SELECT * FROM test.q1_2 where id = 1 ORDER BY arrival_delay ASC LIMIT 10;\n",
    "def Q1_2(parsed):\n",
    "    parsed.map(lambda v: (v[\"UniqueCarrier\"], [v[\"UniqueCarrier\"], float(v[\"ArrDelay\"]), 1])) \\\n",
    "        .reduceByKey(lambda a,b: (a[0], a[1] + b[1], a[2] + b[2])) \\\n",
    "        .map(lambda v: (v[0], (v[1][1], v[1][2]))) \\\n",
    "        .updateStateByKey(update_Q1_2) \\\n",
    "        .map(lambda v: (1, v[1][0] / v[1][1], v[0])) \\\n",
    "        .foreachRDD(process1_2)\n",
    "\n",
    "# ======================================================================================\n",
    "# Question 2.1 - For each airport X, \n",
    "#                rank the top-10 carriers in decreasing order of on-time departure performance from X.\n",
    "# ======================================================================================\n",
    "# CREATE TABLE test.q2_1 (airport text, departure_delay float, carrier text, PRIMARY KEY(airport, departure_delay, carrier));\n",
    "# SELECT * FROM test.q2_1 WHERE airport='SRQ' ORDER BY departure_delay LIMIT 10;\n",
    "def Q2_1(parsed):\n",
    "    parsed.map(lambda v: ('{}_{}'.format(v[\"Origin\"], v[\"UniqueCarrier\"]), (float(v[\"DepDelay\"]), 1))) \\\n",
    "        .reduceByKey(lambda a,b: (a[0] + b[0], a[1] + b[1])) \\\n",
    "        .updateStateByKey(update_Q2) \\\n",
    "        .map(lambda v: (v[0].split(\"_\")[0], (v[0].split(\"_\")[1], v[1][0] / v[1][1]))) \\\n",
    "        .foreachRDD(process2_1)\n",
    "    \n",
    "# ======================================================================================\n",
    "# Question 2.2 - For each source airport X, \n",
    "#                rank the top-10 destination airports in decreasing order of on-time departure performance from X.\n",
    "# ======================================================================================\n",
    "# CREATE TABLE test.q2_2 (airport text, departure_delay float, destination_airport text, PRIMARY KEY(airport, departure_delay, destination_airport));\n",
    "# SELECT * FROM test.q2_2 WHERE airport='SRQ' ORDER BY departure_delay LIMIT 10;\n",
    "def Q2_2(parsed):\n",
    "    parsed.map(lambda v: ('{}_{}'.format(v[\"Origin\"], v[\"Dest\"]), (float(v[\"DepDelay\"]), 1))) \\\n",
    "        .reduceByKey(lambda a,b: (a[0] + b[0], a[1] + b[1])) \\\n",
    "        .updateStateByKey(update_Q2) \\\n",
    "        .map(lambda v: (v[0].split(\"_\")[0], (v[0].split(\"_\")[1], v[1][0] / v[1][1]))) \\\n",
    "        .foreachRDD(process2_2)\n",
    "\n",
    "# ======================================================================================\n",
    "# Question 2.3 - For each source-destination pair X-Y, \n",
    "#                rank the top-10 carriers in decreasing order of on-time arrival performance at Y from X.\n",
    "# ======================================================================================\n",
    "# CREATE TABLE test.q2_3 (airport text, destination_airport text, arrival_delay float, carrier text, PRIMARY KEY((airport, destination_airport), arrival_delay, carrier));\n",
    "# SELECT * FROM test.q2_3 WHERE airport='LGA' and destination_airport='BOS' ORDER BY arrival_delay LIMIT 10;\n",
    "def Q2_3(parsed):\n",
    "    parsed.map(lambda v: ('{}_{}_{}'.format(v[\"Origin\"], v[\"Dest\"], v[\"UniqueCarrier\"]), (float(v[\"ArrDelay\"]), 1))) \\\n",
    "        .reduceByKey(lambda a,b: (a[0] + b[0], a[1] + b[1])) \\\n",
    "        .updateStateByKey(update_Q2) \\\n",
    "        .map(lambda v: (v[0].split(\"_\")[0], v[0].split(\"_\")[1], v[0].split(\"_\")[2], v[1][0] / v[1][1])) \\\n",
    "        .foreachRDD(process2_3)\n",
    "\n",
    "# ======================================================================================\n",
    "# Question 3.2\n",
    "# ======================================================================================\n",
    "\n",
    "def increase_2_days(date):\n",
    "    dt = parser.parse(date)\n",
    "    future_date = dt + datetime.timedelta(days=2)\n",
    "    return future_date.strftime(\"%Y-%m-%d\")\n",
    "\n",
    "def process3_2(time, rdd):\n",
    "    print(\"========= %s =========\" % str(time))\n",
    "\n",
    "    try:\n",
    "        if (not rdd.take(1)):\n",
    "            return\n",
    "        \n",
    "        spark = getSparkSessionInstance(rdd.context.getConf())\n",
    "        rowRdd = rdd.map(lambda w: Row(carrier=w[\"UniqueCarrier\"],\\\n",
    "                                       origin=w[\"Origin\"],\\\n",
    "                                       destination=w[\"Dest\"],\\\n",
    "                                       arrival_delay=float(w[\"ArrDelay\"]),\\\n",
    "                                       date=w[\"FlightDate\"],\\\n",
    "                                       future_date=increase_2_days(w[\"FlightDate\"]),\\\n",
    "                                       time=w[\"CRSDepTime\"]))\n",
    "        wordsDataFrame = spark.createDataFrame(rowRdd)\n",
    "        wordsDataFrame.createOrReplaceTempView(\"flights\")\n",
    "        sqlCommand = \\\n",
    "    \"SELECT a.origin as lag1_origin, \" + \\\n",
    "           \"a.destination as lag1_destination, \" + \\\n",
    "           \"a.carrier as lag1_carrier, \" + \\\n",
    "           \"a.arrival_delay as lag1_arrival_delay, \" + \\\n",
    "           \"a.date as lag1_date, \" + \\\n",
    "           \"a.time as lag1_time, \" + \\\n",
    "           \"b.origin as lag2_origin, \" + \\\n",
    "           \"b.destination as lag2_destination, \" + \\\n",
    "           \"b.carrier as lag2_carrier, \" + \\\n",
    "           \"b.arrival_delay as lag2_arrival_delay, \" + \\\n",
    "           \"b.date as lag2_date, \" + \\\n",
    "           \"b.time as lag2_time \" + \\\n",
    "    \"FROM flights AS a \" + \\\n",
    "    \"INNER JOIN flights AS b ON a.destination = b.origin AND a.future_date=b.date AND a.time<1200 AND b.time>=1200\" \n",
    "        spark.sql(sqlCommand).write\\\n",
    "            .format(\"org.apache.spark.sql.cassandra\")\\\n",
    "            .mode('overwrite')\\\n",
    "            .options(table=\"q3_2\", keyspace=\"test\")\\\n",
    "            .option(\"confirm.truncate\",\"true\")\\\n",
    "            .save()         \n",
    "\n",
    "    except:\n",
    "        print(\"Oops!\",sys.exc_info()[0],\"occured.\")\n",
    "        pass \n",
    "\n",
    "def Q3_2(parsed):\n",
    "    parsed \\\n",
    "        .filter(lambda v: v[\"FlightDate\"].find(\"2008-\") >= 0) \\\n",
    "        .map(lambda v: ('{}_{}_{}_{}' \\\n",
    "            .format(v[\"Origin\"], v[\"Dest\"], v[\"FlightDate\"], \"PM\" if int(v[\"CRSDepTime\"]) > 1200 else \"AM\"), v)) \\\n",
    "        .reduceByKey(lambda a,b: b if float(a[\"ArrDelay\"]) >= float(b[\"ArrDelay\"]) else a) \\\n",
    "        .map(lambda v: v[1]) \\\n",
    "        .foreachRDD(process3_2)\n",
    "    \n",
    "\n",
    "parsed.count().pprint()\n",
    "Q1_1(parsed)\n",
    "# Q1_2(parsed)\n",
    "# Q2_1(parsed)\n",
    "# Q2_2(parsed)\n",
    "# Q2_3(parsed)\n",
    "# Q3_2(parsed)\n",
    "    \n",
    "# Start the stream\n",
    "ssc.start()\n",
    "ssc.awaitTermination()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}