Spark-structured-streaming-from-Kafka.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## To use this notebook\n",
    "\n",
    "Jupyter Notebooks allow you to modify and run the code in this document. To run a section (known as a 'cell',) select it and then use CTRL + ENTER, or select the play button on the toolbar above. Note that each section already has some example output beneath it, so you can see what the results of running a cell will look like.\n",
    "\n",
    "NOTE: You must run each cell in order, from top to bottom. Running cells out of order can result in an error.\n",
    "\n",
    "## Requirements\n",
    "\n",
    "* An Azure Virtual Network\n",
    "* A Spark on HDInsight 3.6 cluster, inside the virtual network\n",
    "* A Kafka on HDInsight cluster, inside the virtual network\n",
    "\n",
    "## Load packages\n",
    "\n",
    "To use Spark structured streaming with Kafka, you must load the spark-sql-kafka package. The version must match the version of both kafka and Spark that you are using. The name of the package contains the versions that it works with. For example, `spark-sql-kafka-0-10_2.11:2.1.0` works with the following versions:\n",
    "\n",
    "* Kafka 0.10\n",
    "* Spark 2.1.0\n",
    "* Scala 2.11\n",
    "\n",
    "Run the next cell to load a package that works with Kafka on HDInsight 3.6, and Spark 2.1 on HDInsight 3.6."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "Current session configs: <tt>{u'kind': 'spark', u'conf': {u'spark.jars.packages': u'org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0', u'spark.jars.excludes': u'org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11'}}</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>Current session?</th></tr><tr><td>2</td><td>application_1504705746198_0006</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://hn1-spark.dcmr0djp5tqejjuyc2l5zdetee.jx.internal.cloudapp.net:8088/proxy/application_1504705746198_0006/\">Link</a></td><td><a target=\"_blank\" href=\"http://10.0.0.4:30060/node/containerlogs/container_1504705746198_0006_01_000001/livy\">Link</a></td><td></td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%configure -f\n",
    "{\n",
    "    \"conf\": {\n",
    "        \"spark.jars.packages\": \"org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0\", \n",
    "        \"spark.jars.excludes\": \"org.scala-lang:scala-reflect,org.apache.spark:spark-tags_2.11\"\n",
    "    }\n",
    "}\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define a schema for the data\n",
    "When reading data from Kafka, the data is provided in the 'value' column. In this example, the data is a JSON document that describes a Tweet. Run the following cell to create a schema for the JSON document structure."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "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>Current session?</th></tr><tr><td>3</td><td>application_1504705746198_0007</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http://hn1-spark.dcmr0djp5tqejjuyc2l5zdetee.jx.internal.cloudapp.net:8088/proxy/application_1504705746198_0007/\">Link</a></td><td><a target=\"_blank\" href=\"http://10.0.0.4:30060/node/containerlogs/container_1504705746198_0007_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n",
      "schema: org.apache.spark.sql.types.StructType = StructType(StructField(created_at,StringType,true), StructField(id,LongType,true), StructField(id_str,StringType,true), StructField(text,StringType,true), StructField(source,StringType,true), StructField(truncated,BooleanType,true), StructField(in_reply_to_status_id,LongType,true), StructField(in_reply_to_status_id_str,StringType,true), StructField(in_reply_to_user_id,LongType,true), StructField(in_reply_to_user_id_str,StringType,true), StructField(in_reply_to_screen_name,StringType,true), StructField(user,StructType(StructField(id,LongType,true), StructField(id_str,StringType,true), StructField(name,StringType,true), StructField(screen_name,StringType,true), StructField(location,StringType,true), StructField(url,StringType,true), StructFi..."
     ]
    }
   ],
   "source": [
    "// Import bits useed for declaring schemas and working with JSON data\n",
    "import org.apache.spark.sql._\n",
    "import org.apache.spark.sql.types._\n",
    "import org.apache.spark.sql.functions._\n",
    "\n",
    "// Define the structure of the Twitter JSON document that is read from Kafka\n",
    "// Note, this isn't pretty, but there's some odd behavior where moving .add to \n",
    "// a new line causes an error.\n",
    "val schema = (new StructType).add(\"created_at\", StringType).add(\"id\", LongType).add(\"id_str\", StringType).add(\"text\", StringType).add(\"source\", StringType).add(\"truncated\", BooleanType).add(\"in_reply_to_status_id\", LongType).add(\"in_reply_to_status_id_str\", StringType).add(\"in_reply_to_user_id\", LongType).add(\"in_reply_to_user_id_str\", StringType).add(\"in_reply_to_screen_name\", StringType).add(\"user\", (new StructType).add(\"id\", LongType)\n",
    "        .add(\"id_str\", StringType)\n",
    "        .add(\"name\", StringType)\n",
    "        .add(\"screen_name\", StringType)\n",
    "        .add(\"location\", StringType)\n",
    "        .add(\"url\", StringType)\n",
    "        .add(\"description\", StringType)\n",
    "        .add(\"protected\", BooleanType)\n",
    "        .add(\"verified\", BooleanType)\n",
    "        .add(\"followers_count\", LongType)\n",
    "        .add(\"friends_count\", LongType)\n",
    "        .add(\"listed_count\", LongType)\n",
    "        .add(\"favourites_count\", LongType)\n",
    "        .add(\"statuses_count\", LongType)\n",
    "        .add(\"created_at\", StringType)\n",
    "        .add(\"utc_offset\", IntegerType)\n",
    "        .add(\"time_zone\", StringType)\n",
    "        .add(\"geo_enabled\", BooleanType)\n",
    "        .add(\"lang\", StringType)\n",
    "        .add(\"contributors_enabled\", BooleanType)\n",
    "        .add(\"is_translator\", BooleanType)\n",
    "        .add(\"profile_background_color\", StringType)\n",
    "        .add(\"profile_background_image_url\", StringType)\n",
    "        .add(\"profile_background_image_url_https\", StringType)\n",
    "        .add(\"profile_background_tile\", BooleanType)\n",
    "        .add(\"profile_link_color\", StringType)\n",
    "        .add(\"profile_sidebar_border_color\", StringType)\n",
    "        .add(\"profile_sidebar_fill_color\", StringType)\n",
    "        .add(\"profile_text_color\", StringType)\n",
    "        .add(\"profile_use_background_image\", BooleanType)\n",
    "        .add(\"profile_image_url\", StringType)\n",
    "        .add(\"profile_image_url_https\", StringType)\n",
    "        .add(\"profile_banner_url\", StringType)\n",
    "        .add(\"default_profile\", BooleanType)\n",
    "        .add(\"default_profile_image\", BooleanType)\n",
    "        .add(\"following\", StringType)\n",
    "        .add(\"follow_request_sent\", StringType)\n",
    "        .add(\"notifications\", StringType)).add(\"geo\", StringType).add(\"coordinates\", StringType).add(\"place\", StringType).add(\"contributors\", StringType).add(\"is_quote_status\", BooleanType).add(\"retweet_count\", LongType).add(\"favorite_count\", LongType).add(\"entities\", (new StructType)\n",
    "        .add(\"hashtags\", ArrayType((new StructType)\n",
    "            .add(\"text\", StringType)\n",
    "            .add(\"indices\", ArrayType(LongType)))).add(\"urls\", ArrayType((new StructType)\n",
    "            .add(\"url\", StringType)\n",
    "            .add(\"expanded_url\", StringType)\n",
    "            .add(\"display_url\", StringType)\n",
    "            .add(\"indices\", ArrayType(LongType))))\n",
    "        .add(\"user_mentions\", ArrayType(StringType))\n",
    "        .add(\"symbols\", ArrayType(StringType))).add(\"favorited\", BooleanType).add(\"retweeted\", BooleanType).add(\"possibly_sensitive\", BooleanType).add(\"filter_level\", StringType).add(\"lang\", StringType).add(\"timestamp_ms\", StringType)\n",
    "\n",
    "// Uncomment to see a tree view of the schema.\n",
    "//schema.printTreeString"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read the data and apply the schema\n",
    "\n",
    "In the following cell, replace `YOUR_KAFKA_BROKER_HOSTS` with the broker hosts for your Kafka cluster. To get the broker host information, use one of the following methods:\n",
    "\n",
    "* From __Bash__ or other Unix shell:\n",
    "\n",
    "    ```bash\n",
    "curl -u admin:$PASSWORD -G \"https://$CLUSTERNAME.azurehdinsight.net/api/v1/clusters/$CLUSTERNAME/services/KAFKA/components/KAFKA_BROKER\" | jq -r '[\"\\(.host_components[].HostRoles.host_name):9092\"] | join(\",\")' | cut -d',' -f1,2\n",
    "    ```\n",
    "    \n",
    "    Note: This assumes that `$PASSWORD` is set to the password for your HDInsight cluster admin, and that `$CLUSTERNAME` is set to the name of the cluster.\n",
    "\n",
    "* From __Azure Powershell__:\n",
    "\n",
    "    ```powershell\n",
    "$creds = Get-Credential -UserName \"admin\" -Message \"Enter the HDInsight login\"\n",
    "$clusterName = Read-Host -Prompt \"Enter the Kafka cluster name\"\n",
    "$resp = Invoke-WebRequest -Uri \"https://$clusterName.azurehdinsight.net/api/v1/clusters/$clusterName/services/KAFKA/components/KAFKA_BROKER\" `\n",
    "    -Credential $creds\n",
    "$respObj = ConvertFrom-Json $resp.Content\n",
    "$brokerHosts = $respObj.host_components.HostRoles.host_name[0..1]\n",
    "($brokerHosts -join \":9092,\") + \":9092\"\n",
    "    ```\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Finished configuring the fields we want to select from the stream."
     ]
    }
   ],
   "source": [
    "// Read from the Kafka stream source\n",
    "val kafka = spark.readStream.format(\"kafka\").option(\"kafka.bootstrap.servers\", \"YOUR_KAFKA_BROKER_HOSTS\").option(\"subscribe\", \"tweets\").option(\"startingOffsets\",\"earliest\").load()\n",
    "\n",
    "/* Select the following columns from the Kafka data:\n",
    "   * value - the JSON data for a tweet\n",
    "   Use from_json to apply the schema and store the schematized data in the 'tweet' column\n",
    "*/\n",
    "val tweetData=kafka.select(\n",
    "    from_json(col(\"value\").cast(\"string\"), schema) as \"tweet\")\n",
    "\n",
    "// There's a lot of data in the Twitter JSON object. Just grab the tweet ID, user name, and text\n",
    "val tweetText=tweetData.select(\"tweet.id\",\n",
    "                               \"tweet.user.name\",\n",
    "                               \"tweet.text\")\n",
    "\n",
    "println(\"Finished configuring the fields we want to select from the stream.\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Process the stream\n",
    "\n",
    "To start processing the stream, write it to a sink. Run the following cell to write the data to the console (cell output). This cell runs for 30 seconds, then displays the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------------------------------------\n",
      "Batch: 0\n",
      "-------------------------------------------\n",
      "+------------------+--------------------+--------------------+\n",
      "|                id|                name|                text|\n",
      "+------------------+--------------------+--------------------+\n",
      "|905441423759171585|       Lady Alphonse|16 años :(\n",
      "¿Pero ...|\n",
      "|905441422794309632|         Ryukyu-blue|RT @takaatsurit: ...|\n",
      "|905441423033372673|         aly merrill|RT @priscillux: W...|\n",
      "|905441422656057345|            Katlyn☮️|RT @INDIEWASHERE:...|\n",
      "|905441422941323264|                  Mo|RT @JBrewerBoston...|\n",
      "|905441422915977216|         #DefendDACA|RT @SkyNews: Hurr...|\n",
      "|905441423322898433|             Rhya🌷✨|RT @flor_demaga: ...|\n",
      "|905441422534434816|      Franklin Lopez|Directorio de tel...|\n",
      "|905441423859863552|             Laila|RT @RCI_GP: [IRMA...|\n",
      "|905441422962253824|            Lanna 🌸|RT @txflxn: Freak...|\n",
      "|905441422677090305|     Finlay Copeland|RT @PopeQuanPaul:...|\n",
      "|905441422169571328|         FlamingFran|RT @Complex: Seve...|\n",
      "|905441424488923137|           Jul⚽🏈⚾🏁|RT @EricHolthaus:...|\n",
      "|905441423553511424|                Yama|RT @everlasting50...|\n",
      "|905441423188787200|       RAMIRO ESCOTO|asi el paso de #I...|\n",
      "|905441423348174849|  Tina \"Anti-GOP\" E.|Rush Limbaugh Cla...|\n",
      "|905441423180394498|                 Lal|I'm at More Coffe...|\n",
      "|905441425461936128|    Princess La Lani|RT @AMHQ: Broward...|\n",
      "|905441423981510656|DeaconBlue #MAGA🇺🇸|RT @gearmeister: ...|\n",
      "|905441424979542016|         Live Purple|It's #NationalRea...|\n",
      "+------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 1\n",
      "-------------------------------------------\n",
      "+------------------+--------------------+--------------------+\n",
      "|                id|                name|                text|\n",
      "+------------------+--------------------+--------------------+\n",
      "|905446383326568450|          wxthehague|The Hague Weather...|\n",
      "|905446383058132992|           Jeni 🐾🐕|RT @StephenKing: ...|\n",
      "|905446383276216320|  Bradley Scott 🇺🇸|RT @bradcrain: On...|\n",
      "|905446382869336064|   (((Antagoniste)))|Trump s'est torch...|\n",
      "|905446387407405056|   giselle elisabeth|RT @priscillux: W...|\n",
      "|905446383037054977|              tuszyb|RT @AdamSchiffCA:...|\n",
      "|905446383339089921|             Kaitlan|RT @BuzzFeedNews:...|\n",
      "|905446387101392896|          Antoinette|RT @RogueSNRadvis...|\n",
      "|905446387164356608|     Matt Serwe KETV|If you've seen th...|\n",
      "|905446382986825728|             Giselle|RT @PopeQuanPaul:...|\n",
      "|905446384622546944|                  MT|RT @LilFloFrmDaDa...|\n",
      "|905446384068853761|          Lewis Cook|@sahilkapur @Watc...|\n",
      "|905446385117519877|       Brina Starler|Very calming and ...|\n",
      "|905446387298451457|        Jeff Huggins|RT @Pismo_B: Real...|\n",
      "|905446387067879426| Liberal News - 2012|CNN:  Blue states...|\n",
      "|905446383355912193|                 Lo'|If the mountains ...|\n",
      "|905446382856699908|Sell More and Better|RT @RaulGilo: #We...|\n",
      "|905446386837147648|        Robin Wesley|RT @UNITEDWEDREAM...|\n",
      "|905446387059494913|            ☇B. B.🍺|RT @LisaLamb84937...|\n",
      "|905446387189415936|            skrbelly|RT @ditzkoff: Tru...|\n",
      "+------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "-------------------------------------------\n",
      "Batch: 2\n",
      "-------------------------------------------\n",
      "res23: Boolean = false"
     ]
    }
   ],
   "source": [
    "// Start writing the stream to the console. Use a timeout so that control is returned to the notebook.\n",
    "tweetText.writeStream.format(\"console\").start.awaitTermination(30000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "// Write the stream to HDInsight storage\n",
    "tweetText.writeStream.format(\"parquet\").option(\"path\",\"/example/tweets\").option(\"checkpointLocation\", \"/checkpoint\").start.awaitTermination(60000)\n",
    "// Data is written to WASB or ADL at `/example/tweets`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Spark",
   "language": "",
   "name": "sparkkernel"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}