3_advanced/1_Stream-Stream Join Demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Stream-Stream Join Demo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Join Operations\n",
    "\n",
    "Finally, its worth highlighting how easily you can perform different kinds of joins in Spark Streaming.\n",
    "\n",
    "### Stream-stream joins\n",
    "\n",
    "Streams can be very easily joined with other streams.\n",
    "```python\n",
    "stream1 = ...\n",
    "stream2 = ...\n",
    "joinedStream = stream1.join(stream2)\n",
    "```\n",
    "Here, in each batch interval, the RDD generated by `stream1` will be joined with the RDD generated by `stream2`. You can also do `leftOuterJoin`, `rightOuterJoin`, `fullOuterJoin`. Furthermore, it is often very useful to do joins over windows of the streams. That is pretty easy as well.\n",
    "```python\n",
    "windowedStream1 = stream1.window(20)\n",
    "windowedStream2 = stream2.window(60)\n",
    "joinedStream = windowedStream1.join(windowedStream2)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Demo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "# TODO: your path will likely not have 'matthew' in it. Change it to reflect your path.\n",
    "findspark.init('/home/matthew/spark-2.1.0-bin-hadoop2.7')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "from pyspark.streaming import StreamingContext\n",
    "from pprint import pprint\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext()\n",
    "ssc = StreamingContext(sc, 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[PythonRDD[5] at RDD at PythonRDD.scala:48,\n",
      " PythonRDD[6] at RDD at PythonRDD.scala:48,\n",
      " PythonRDD[7] at RDD at PythonRDD.scala:48,\n",
      " PythonRDD[8] at RDD at PythonRDD.scala:48,\n",
      " PythonRDD[9] at RDD at PythonRDD.scala:48]\n"
     ]
    }
   ],
   "source": [
    "rdd_queue = []\n",
    "for i in range(5): \n",
    "    rdd_data = range(1000)\n",
    "    rdd = ssc.sparkContext.parallelize(rdd_data)\n",
    "    rdd_queue.append(rdd)\n",
    "pprint(rdd_queue)\n",
    "\n",
    "# Creating queue stream # 1\n",
    "ds1 = ssc.queueStream(rdd_queue).map(lambda x: (x % 10, 1)).window(4).reduceByKey(lambda v1,v2:v1+v2)\n",
    "ds1.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating queue stream # 2\n",
    "ds2 = ssc.queueStream(rdd_queue).map(lambda x: (x % 5, 1)).window(windowDuration=20).reduceByKey(lambda v1,v2:v1+v2)\n",
    "ds2.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Crossing the Streams\n",
    "joinedStream = ds1.join(ds2)\n",
    "joinedStream.pprint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:45\n",
      "-------------------------------------------\n",
      "(0, 100)\n",
      "(8, 100)\n",
      "(2, 100)\n",
      "(4, 100)\n",
      "(6, 100)\n",
      "(1, 100)\n",
      "(3, 100)\n",
      "(9, 100)\n",
      "(5, 100)\n",
      "(7, 100)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:45\n",
      "-------------------------------------------\n",
      "(0, 200)\n",
      "(2, 200)\n",
      "(4, 200)\n",
      "(1, 200)\n",
      "(3, 200)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:45\n",
      "-------------------------------------------\n",
      "(0, (100, 200))\n",
      "(2, (100, 200))\n",
      "(4, (100, 200))\n",
      "(1, (100, 200))\n",
      "(3, (100, 200))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:46\n",
      "-------------------------------------------\n",
      "(0, 200)\n",
      "(8, 200)\n",
      "(2, 200)\n",
      "(4, 200)\n",
      "(6, 200)\n",
      "(1, 200)\n",
      "(3, 200)\n",
      "(9, 200)\n",
      "(5, 200)\n",
      "(7, 200)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:46\n",
      "-------------------------------------------\n",
      "(0, 400)\n",
      "(2, 400)\n",
      "(4, 400)\n",
      "(1, 400)\n",
      "(3, 400)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:46\n",
      "-------------------------------------------\n",
      "(0, (200, 400))\n",
      "(2, (200, 400))\n",
      "(4, (200, 400))\n",
      "(1, (200, 400))\n",
      "(3, (200, 400))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:47\n",
      "-------------------------------------------\n",
      "(0, 300)\n",
      "(8, 300)\n",
      "(2, 300)\n",
      "(4, 300)\n",
      "(6, 300)\n",
      "(1, 300)\n",
      "(3, 300)\n",
      "(9, 300)\n",
      "(5, 300)\n",
      "(7, 300)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:47\n",
      "-------------------------------------------\n",
      "(0, 600)\n",
      "(2, 600)\n",
      "(4, 600)\n",
      "(1, 600)\n",
      "(3, 600)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:47\n",
      "-------------------------------------------\n",
      "(0, (300, 600))\n",
      "(2, (300, 600))\n",
      "(4, (300, 600))\n",
      "(1, (300, 600))\n",
      "(3, (300, 600))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:48\n",
      "-------------------------------------------\n",
      "(0, 400)\n",
      "(8, 400)\n",
      "(2, 400)\n",
      "(4, 400)\n",
      "(6, 400)\n",
      "(1, 400)\n",
      "(3, 400)\n",
      "(9, 400)\n",
      "(5, 400)\n",
      "(7, 400)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:48\n",
      "-------------------------------------------\n",
      "(0, 800)\n",
      "(2, 800)\n",
      "(4, 800)\n",
      "(1, 800)\n",
      "(3, 800)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:48\n",
      "-------------------------------------------\n",
      "(0, (400, 800))\n",
      "(2, (400, 800))\n",
      "(4, (400, 800))\n",
      "(1, (400, 800))\n",
      "(3, (400, 800))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:49\n",
      "-------------------------------------------\n",
      "(0, 400)\n",
      "(8, 400)\n",
      "(2, 400)\n",
      "(4, 400)\n",
      "(6, 400)\n",
      "(1, 400)\n",
      "(3, 400)\n",
      "(9, 400)\n",
      "(5, 400)\n",
      "(7, 400)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:49\n",
      "-------------------------------------------\n",
      "(0, 1000)\n",
      "(2, 1000)\n",
      "(4, 1000)\n",
      "(1, 1000)\n",
      "(3, 1000)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:49\n",
      "-------------------------------------------\n",
      "(0, (400, 1000))\n",
      "(2, (400, 1000))\n",
      "(4, (400, 1000))\n",
      "(1, (400, 1000))\n",
      "(3, (400, 1000))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:50\n",
      "-------------------------------------------\n",
      "(0, 300)\n",
      "(8, 300)\n",
      "(2, 300)\n",
      "(4, 300)\n",
      "(6, 300)\n",
      "(1, 300)\n",
      "(3, 300)\n",
      "(9, 300)\n",
      "(5, 300)\n",
      "(7, 300)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:50\n",
      "-------------------------------------------\n",
      "(0, 1000)\n",
      "(2, 1000)\n",
      "(4, 1000)\n",
      "(1, 1000)\n",
      "(3, 1000)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:50\n",
      "-------------------------------------------\n",
      "(0, (300, 1000))\n",
      "(2, (300, 1000))\n",
      "(4, (300, 1000))\n",
      "(1, (300, 1000))\n",
      "(3, (300, 1000))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:51\n",
      "-------------------------------------------\n",
      "(0, 200)\n",
      "(8, 200)\n",
      "(2, 200)\n",
      "(4, 200)\n",
      "(6, 200)\n",
      "(1, 200)\n",
      "(3, 200)\n",
      "(9, 200)\n",
      "(5, 200)\n",
      "(7, 200)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:51\n",
      "-------------------------------------------\n",
      "(0, 1000)\n",
      "(2, 1000)\n",
      "(4, 1000)\n",
      "(1, 1000)\n",
      "(3, 1000)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:51\n",
      "-------------------------------------------\n",
      "(0, (200, 1000))\n",
      "(2, (200, 1000))\n",
      "(4, (200, 1000))\n",
      "(1, (200, 1000))\n",
      "(3, (200, 1000))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:52\n",
      "-------------------------------------------\n",
      "(0, 100)\n",
      "(8, 100)\n",
      "(2, 100)\n",
      "(4, 100)\n",
      "(6, 100)\n",
      "(1, 100)\n",
      "(3, 100)\n",
      "(9, 100)\n",
      "(5, 100)\n",
      "(7, 100)\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:52\n",
      "-------------------------------------------\n",
      "(0, 1000)\n",
      "(2, 1000)\n",
      "(4, 1000)\n",
      "(1, 1000)\n",
      "(3, 1000)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ssc.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:52\n",
      "-------------------------------------------\n",
      "(0, (100, 1000))\n",
      "(2, (100, 1000))\n",
      "(4, (100, 1000))\n",
      "(1, (100, 1000))\n",
      "(3, (100, 1000))\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:53\n",
      "-------------------------------------------\n",
      "\n",
      "-------------------------------------------\n",
      "Time: 2018-03-01 22:03:53\n",
      "-------------------------------------------\n",
      "(0, 1000)\n",
      "(2, 1000)\n",
      "(4, 1000)\n",
      "(1, 1000)\n",
      "(3, 1000)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ssc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reference\n",
    "1. https://spark.apache.org/docs/latest/streaming-programming-guide.html#join-operations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " "
   ]
  }
 ],
 "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}