{ "metadata": { "kernelspec": { "name": "SQL", "display_name": "SQL", "language": "sql" }, "language_info": { "name": "sql", "version": "" } }, "nbformat_minor": 2, "nbformat": 4, "cells": [ { "cell_type": "markdown", "source": [ "

\r\n", "\r\n", "

\r\n", "\r\n", "# **Twitter Streaming with SQL Server & Spark**\r\n", "\r\n", "In this notebook, we will go through the process of using Spark to stream tweets from the Twitter API, and then stream the resulting data into the SQL Server data pool. To stream the data to SQL Server data pools, we use the MsSQL-Spark connector. Once the data is in the data pool, we will perform queries on it using T-SQL or the Spark-SQL connector. \r\n", "\r\n", "## **Steps**\r\n", "1. [Create a Twitter Developer Account](https://developer.twitter.com/en/apply-for-access.html).\r\n", "2. Setup\r\n", " 1. Create 'TwitterData' database.\r\n", " 2. Create an External Data Source 'TweetsDataSource'. \r\n", " 3. Create an External Table 'Tweets'.\r\n", " 4. Create a user account with permissions for data pools, created datasource and the external table. Refer [here](https://github.com/microsoft/sql-server-samples/blob/master/samples/features/sql-big-data-cluster/spark/data-virtualization/mssql_spark_connector_user_creation.ipynb) \r\n", " 4. Change kernel from \"SQL\" to \"Spark | Scala\".\r\n", " 5. Import packages.\r\n", " 6. Enter required parameters.\r\n", "3. Define and create a TwitterStream object.\r\n", "4. Start the TwitterStream.\r\n", "5. Validate streaming data.\r\n", "6. Stream data into SQL Server data pool.\r\n", "7. Query the data from the data pool external table using T-SQL or the Spark-SQL connector.\r\n", "8. Stop the TwitterStream" ], "metadata": { "azdata_cell_guid": "edf821ac-d627-4d6e-8c7c-7ee918f072f3" } }, { "cell_type": "markdown", "source": [ "## **1. Create a Twitter Developer Account**\n", "[Create a Twitter Developer Account](https://developer.twitter.com/en/apply-for-access.html) and enter the credentials in the Setup section below. These credentials will be used to authenticate the application with Twitter and allow you to stream data from the platform." ], "metadata": { "azdata_cell_guid": "d3a9df62-793b-4415-9574-4edfccb205ba" } }, { "cell_type": "markdown", "source": [ "## **2. Setup**\n", "1. Create a database in the SQL Server master instance named 'TwitterData'.\n", "2. Create an External Data Source to the Data Pool named 'TweetsDataSource'.\n", "3. Create an External Table in the Data Pool named 'Tweets'.\n", "4. Change the Kernel from \"SQL\" to \"Spark | Scala\".\n", "5. Import Java packages.\n", "6. Specify setup parameters" ], "metadata": { "azdata_cell_guid": "514963d4-c9eb-42a7-bd81-c6735f79d647" } }, { "cell_type": "markdown", "source": [ "### **2.1 Create TwitterData Database and Retrieve Hostname**" ], "metadata": { "azdata_cell_guid": "196dd8cd-e4d7-45c7-b6f8-21dde6c75ae7" } }, { "cell_type": "code", "source": [ "USE master\n", "IF EXISTS(select * from sys.databases where name='TwitterData')\n", "DROP DATABASE TwitterData;\n", "GO\n", "CREATE DATABASE TwitterData;\n", "GO\n", "USE TwitterData;\n", "GO" ], "metadata": { "azdata_cell_guid": "abf4ebf4-c8fc-4316-a300-2c64bd8f4592" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "Use the following cell to check your server name. If it returns ```master-0```, your hostname will be ```master-0.master-svc```. You will define this value in the Parameters section." ], "metadata": { "azdata_cell_guid": "02e6cd9a-ef88-419c-afa0-e6a666f1070e" } }, { "cell_type": "code", "source": [ "SELECT @@Servername;" ], "metadata": { "azdata_cell_guid": "0827942d-a223-4ba8-8168-2fb57a4a2488" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "### **2.2 Create External Data Source 'TweetsDataSource'**" ], "metadata": { "azdata_cell_guid": "03542af4-1e39-4049-a982-a44fce4cebd4" } }, { "cell_type": "code", "source": [ "USE TwitterData\n", "GO\n", "\n", "IF NOT EXISTS(SELECT * FROM sys.external_data_sources WHERE name = 'TweetsDataSource')\n", " CREATE EXTERNAL DATA SOURCE TweetsDataSource\n", " WITH (LOCATION = 'sqldatapool://controller-svc/default');" ], "metadata": { "azdata_cell_guid": "b01e9faf-d701-4a5e-95a3-7afb66b1249b" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "### **2.3 Create External Table 'Tweets'**" ], "metadata": { "azdata_cell_guid": "a2576ce9-bd62-4138-937c-f5ccdfe0834e" } }, { "cell_type": "code", "source": [ "IF NOT EXISTS(SELECT * FROM sys.external_tables WHERE name = 'Tweets')\n", " CREATE EXTERNAL TABLE [Tweets]\n", " (\"screen_name\" NVARCHAR(MAX), \"createdAt\" DATETIME , \"num_followers\" BIGINT, \"text\" NVARCHAR(MAX))\n", " WITH\n", " (\n", " DATA_SOURCE = TweetsDataSource,\n", " DISTRIBUTION = ROUND_ROBIN\n", " );" ], "metadata": { "azdata_cell_guid": "e80447c6-92a8-459f-aa17-517b89bd5fed" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "### **2.4 Change the kernel from \"SQL\" to \"Spark | Scala\"**\n", "At the top of the editor, click the Kernel dropdown menu and change the kernel from \"SQL\" to \"Spark | Scala\". This will update the notebook language, and allow you to proceed with the next steps." ], "metadata": { "azdata_cell_guid": "9cf2a253-f6eb-4125-8938-12254ded05a7" } }, { "cell_type": "markdown", "source": [ "### **2.5 Import packages**" ], "metadata": { "azdata_cell_guid": "04406211-4b11-4be8-b0da-e8ade7e6bdfc" } }, { "cell_type": "code", "source": [ "import java.io.{BufferedReader, File, FileNotFoundException, InputStream, InputStreamReader}\r\n", "import java.net.URLEncoder\r\n", "import java.nio.charset.StandardCharsets\r\n", "import java.util.Base64\r\n", "import javax.crypto.Mac\r\n", "import javax.crypto.spec.SecretKeySpec\r\n", "import scala.collection.JavaConverters._\r\n", "import org.apache.commons.io.IOUtils\r\n", "import org.apache.http.client.methods.HttpGet\r\n", "import org.apache.http.impl.client.CloseableHttpClient\r\n", "import org.apache.http.impl.client.HttpClients\r\n", "import org.apache.hadoop.conf.Configuration\r\n", "import org.apache.hadoop.fs.FileSystem\r\n", "import org.apache.hadoop.fs.Path\r\n", "import java.io.PrintWriter\r\n", "import org.apache.spark.sql.{SparkSession, SaveMode, Row, DataFrame}" ], "metadata": { "azdata_cell_guid": "fb2fbadd-c710-4a47-a494-2ecc5db0f375" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "### **2.6 Parameters**\r\n", "Enter the required parameters for the Spark streaming job to connect to SQL Server.\r\n", "\r\n", "In this example, the connection is made from Spark to the SQL Server master instance using the internal DNS name (Ex: master-0.master-svc) and port (1433). Alternatively, and especially if you are using a highly available Always On Availability Group, you can connect to the Kubernetes service that exposes the primary node of the Always On Availability Group.\r\n", "\r\n", "#### Parameters needed to create Twitter stream\r\n", "- Twitter API authentication keys\r\n", "\r\n", "#### Parameters needed for Spark-SQL connector:\r\n", "- user\r\n", "- password\r\n", "\r\n", "#### Optional parameters:\r\n", "- hostname\r\n", "- port \r\n", "- CSV schema\r\n", "- Source directory location\r\n", "- Set path in hdfs to store tweets\r\n", "- Set saving interval for file creation \r\n", "- Set Twitter filters\r\n", "" ], "metadata": { "azdata_cell_guid": "73fad3ec-a900-4398-a710-990f2c30e0f3" } }, { "cell_type": "code", "source": [ "// Twitter app autnentication keys\r\n", "val consumerKey = \"\"\r\n", "val consumerSecret = \"\"\r\n", "val accessToken = \"\"\r\n", "val accessTokenSecret = \"\"\r\n", "\r\n", "// Provide username/password to SQL Server master instance\r\n", "// Refer mssql_spark_connector_user_creation.ipynb in samples repo on how to create this user.\r\n", "val user = \"user\"\r\n", "val password = \"password\"\r\n", "\r\n", "// Spark-SQL connector parameters\r\n", "val hostname = \"master-0.master-svc\"\r\n", "val port = 1433\r\n", "val database = \"TwitterData\"\r\n", "val url = s\"jdbc:sqlserver://${hostname}:${port};database=${database};user=${user};password=${password};\"\r\n", "val dbtable = \"Tweets\"\r\n", "val datasource_name = \"TweetsDataSource\"\r\n", "\r\n", "// Twitter stream object parameters\r\n", "val filters = Array(\"tennis\", \"nadal\", \"federer\", \"murray\", \"djokovic\")\r\n", "val path = \"/user/twitter/\"\r\n", "val savingInterval = 2000" ], "metadata": { "azdata_cell_guid": "3eaece45-6db5-4eab-816c-bdadfffe26ee" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "## **3. Define and Create TwitterStream object**" ], "metadata": { "azdata_cell_guid": "3bbc4717-16d7-42ad-9fc7-f999edc00734" } }, { "cell_type": "code", "source": [ "class TwitterStream(\r\n", " consumerKey: String,\r\n", " consumerSecret: String,\r\n", " accessToken: String,\r\n", " accessTokenSecret: String,\r\n", " path: String,\r\n", " savingInterval: Long,\r\n", " filters: Array[String]) {\r\n", " \r\n", " private val threadName = \"tweet-downloader\"\r\n", " \r\n", " {\r\n", " val hasActiveStream = Thread.getAllStackTraces().keySet().asScala.map(_.getName).contains(threadName)\r\n", " if (hasActiveStream) {\r\n", " throw new RuntimeException(\r\n", " \"There is already an active stream that writes tweets to the configured path. \" +\r\n", " \"Please stop the existing stream first (using twitterStream.stop()).\")\r\n", " }\r\n", " }\r\n", " \r\n", " @volatile private var thread: Thread = null\r\n", " @volatile private var isStopped = false\r\n", " @volatile var isDownloading = false\r\n", " @volatile var exception: Throwable = null\r\n", "\r\n", " private var httpclient: CloseableHttpClient = null\r\n", " private var input: InputStream = null\r\n", " private var httpGet: HttpGet = null\r\n", " \r\n", " private def encode(string: String): String = {\r\n", " URLEncoder.encode(string, StandardCharsets.UTF_8.name)\r\n", " }\r\n", "\r\n", " def start(): Unit = synchronized {\r\n", " isDownloading = false\r\n", " isStopped = false\r\n", " thread = new Thread(threadName) {\r\n", " override def run(): Unit = {\r\n", " httpclient = HttpClients.createDefault()\r\n", " try {\r\n", " requestStream(httpclient)\r\n", " } catch {\r\n", " case e: Throwable => exception = e\r\n", " } finally {\r\n", " //TwitterStream.this.stop()\r\n", " }\r\n", " }\r\n", " }\r\n", " thread.start()\r\n", " }\r\n", "\r\n", " private def requestStream(httpclient: CloseableHttpClient): Unit = {\r\n", " val url = \"https://stream.twitter.com/1.1/statuses/filter.json\"\r\n", " val timestamp = System.currentTimeMillis / 1000\r\n", " val nonce = timestamp + scala.util.Random.nextInt\r\n", " val oauthNonce = nonce.toString\r\n", " val oauthTimestamp = timestamp.toString\r\n", "\r\n", " val oauthHeaderParams = List(\r\n", " \"oauth_consumer_key\" -> encode(consumerKey),\r\n", " \"oauth_signature_method\" -> encode(\"HMAC-SHA1\"),\r\n", " \"oauth_timestamp\" -> encode(oauthTimestamp),\r\n", " \"oauth_nonce\" -> encode(oauthNonce),\r\n", " \"oauth_token\" -> encode(accessToken),\r\n", " \"oauth_version\" -> \"1.0\"\r\n", " )\r\n", " val requestParams = List(\r\n", " \"track\" -> encode(filters.mkString(\",\"))\r\n", " )\r\n", "\r\n", " val parameters = (oauthHeaderParams ++ requestParams).sortBy(_._1).map(pair => s\"\"\"${pair._1}=${pair._2}\"\"\").mkString(\"&\")\r\n", " val base = s\"GET&${encode(url)}&${encode(parameters)}\"\r\n", " val oauthBaseString: String = base.toString\r\n", " val signature = generateSignature(oauthBaseString)\r\n", " val oauthFinalHeaderParams = oauthHeaderParams ::: List(\"oauth_signature\" -> encode(signature))\r\n", " val authHeader = \"OAuth \" + ((oauthFinalHeaderParams.sortBy(_._1).map(pair => s\"\"\"${pair._1}=\"${pair._2}\"\"\"\")).mkString(\", \"))\r\n", "\r\n", " httpGet = new HttpGet(s\"https://stream.twitter.com/1.1/statuses/filter.json?${requestParams.map(pair => s\"\"\"${pair._1}=${pair._2}\"\"\").mkString(\"&\")}\")\r\n", " httpGet.addHeader(\"Authorization\", authHeader)\r\n", " println(\"Downloading tweets!\")\r\n", " val response = httpclient.execute(httpGet)\r\n", " val entity = response.getEntity()\r\n", " input = entity.getContent()\r\n", " if (response.getStatusLine.getStatusCode != 200) {\r\n", " throw new RuntimeException(IOUtils.toString(input, StandardCharsets.UTF_8))\r\n", " }\r\n", " isDownloading = true\r\n", " val reader = new BufferedReader(new InputStreamReader(input, StandardCharsets.UTF_8))\r\n", " var line: String = null\r\n", " var lineno = 1\r\n", " line = reader.readLine()\r\n", " var lastSavingTime = System.currentTimeMillis()\r\n", " val s = new StringBuilder()\r\n", " \r\n", " val conf = new Configuration()\r\n", " val fs= FileSystem.get(conf)\r\n", " \r\n", " while (line != null && !isStopped) {\r\n", " lineno += 1\r\n", " line = reader.readLine()\r\n", " s.append(line + \"\\n\")\r\n", " val now = System.currentTimeMillis()\r\n", " if (now - lastSavingTime >= savingInterval) {\r\n", " \r\n", " val df = spark.read.json(spark.sparkContext.parallelize(Seq(s.toString)))\r\n", " df.write.json(path + now.toString)\r\n", " \r\n", " lastSavingTime = now\r\n", " s.clear()\r\n", " }\r\n", " }\r\n", " }\r\n", "\r\n", " private def generateSignature(data: String): String = {\r\n", " val mac = Mac.getInstance(\"HmacSHA1\")\r\n", " val oauthSignature = encode(consumerSecret) + \"&\" + encode(accessTokenSecret)\r\n", " val spec = new SecretKeySpec(oauthSignature.getBytes, \"HmacSHA1\")\r\n", " mac.init(spec)\r\n", " val byteHMAC = mac.doFinal(data.getBytes)\r\n", " return Base64.getEncoder.encodeToString(byteHMAC)\r\n", " }\r\n", "\r\n", " def stop(): Unit = synchronized {\r\n", " isStopped = true\r\n", " isDownloading = false\r\n", " try {\r\n", " if (httpGet != null) {\r\n", " httpGet.abort()\r\n", " httpGet = null\r\n", " }\r\n", " if (input != null) {\r\n", " input.close()\r\n", " input = null\r\n", " }\r\n", " if (httpclient != null) {\r\n", " httpclient.close()\r\n", " httpclient = null\r\n", " }\r\n", " if (thread != null) {\r\n", " thread.interrupt()\r\n", " thread = null\r\n", " }\r\n", " } catch {\r\n", " case _: Throwable =>\r\n", " }\r\n", " }\r\n", "}\r\n", "println(\"class defined\")" ], "metadata": { "azdata_cell_guid": "aa9b5a2d-4f00-4cb9-8e97-7e5d1bcf78ad" }, "outputs": [], "execution_count": 0 }, { "cell_type": "code", "source": [ "val twitterStream = new TwitterStream(consumerKey, consumerSecret, accessToken, accessTokenSecret, path, savingInterval, filters)" ], "metadata": { "azdata_cell_guid": "b109f960-9fc3-4dbd-869f-996bcae6c294" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "## **4. Start TwitterStream**" ], "metadata": { "azdata_cell_guid": "12f79299-4ce3-4154-8175-0acf2735f068" } }, { "cell_type": "code", "source": [ "twitterStream.start()\r\n", "\r\n", "if (twitterStream.exception != null) { throw twitterStream.exception }" ], "metadata": { "azdata_cell_guid": "f846a370-c9be-499e-8d96-a769e0ffd862" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "## **5. Validate streaming data**\r\n", "\r\n", "Refresh the /user/twitter directory in HDFS and you should see some directories containing data files with the Tweet data." ], "metadata": { "azdata_cell_guid": "ffb3e9d0-0dfc-4060-bb5b-4520958dd890" } }, { "cell_type": "markdown", "source": [ "## **6. Stream data into SQL Server data pool**\r\n", "\r\n", "Now, we can start another job to stream the incoming data into the SQL Server data pool. We will first create the Tweet Dataframe, and then write the data from the dataframe to an external table over the data pool using the Spark-SQL connector." ], "metadata": { "azdata_cell_guid": "0ffd027f-95e4-4718-b806-d4acebbe4874" } }, { "cell_type": "markdown", "source": [ "### **Create Tweet Dataframe**" ], "metadata": { "azdata_cell_guid": "500644d5-d1a2-4a78-96d1-e58fa87e1881" } }, { "cell_type": "code", "source": [ "val tweets = spark.read.json(path + \"*\")\r\n", "val tweets_schema = tweets.schema\r\n", "\r\n", "val tweetStream = spark.readStream.\r\n", "|schema(tweets_schema).\r\n", "|json(path + \"*\").\r\n", "|filter($\"lang\" === \"en\").\r\n", "|withColumn(\"screen_name\", $\"user.screen_name\").\r\n", "|withColumn(\"num_followers\", $\"user.followers_count\").\r\n", "|withColumn(\"createdAt\", from_utc_timestamp(from_unixtime(unix_timestamp($\"created_at\", \"EEE MMM dd HH:mm:ss ZZZZ yyyy\")),\"EST\")).\r\n", "|select(\"screen_name\",\"createdAt\",\"num_followers\", \"text\") " ], "metadata": { "azdata_cell_guid": "d30d7aad-b449-440c-8211-15d73ac8b10a" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "### **Write data to an external table using the Spark-SQL Connector**" ], "metadata": { "azdata_cell_guid": "b4f3476f-ae35-420c-96cd-cf0a9f131c46" } }, { "cell_type": "code", "source": [ "val query = tweetStream.writeStream.outputMode(\"append\").foreachBatch{ (batchDF: DataFrame, batchId: Long) => \r\n", " batchDF.write\r\n", " .format(\"com.microsoft.sqlserver.jdbc.spark\")\r\n", " .mode(\"append\")\r\n", " .option(\"url\", url)\r\n", " .option(\"dbtable\", dbtable)\r\n", " .option(\"user\", user)\r\n", " .option(\"password\", password)\r\n", " .option(\"dataPoolDataSource\",datasource_name).save()\r\n", " }.start()\r\n", "query.processAllAvailable()\r\n", "//query.awaitTermination(40000)" ], "metadata": { "azdata_cell_guid": "eb0690d9-7705-408b-906f-314dc9456242" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "## **7. Query the data from the data pool external table using T-SQL or the Spark-SQL connector**\r\n", "Now, you are streaming data from the source HDFS directory to the data pool table. An external table has been created in the targeted database specified above. You can view the table in the explorer tree and query it using T-SQL. \r\n", "\r\n", "If you want to view the current count of records in the external table, use the code below that uses the Spark-SQL connector to query data from SQL Server into a data frame.\r\n", "\r\n", "You can continue to add files to the /user/twitter directory to see that the Spark-SQL connector automatically picks up new records and adds them to the data pool table." ], "metadata": { "azdata_cell_guid": "b00a4b24-fab5-4315-bae5-965b2beba881" } }, { "cell_type": "code", "source": [ "def df_read(dbtable: String,\r\n", " url: String,\r\n", " dataPoolDataSource: String=\"\"): DataFrame = {\r\n", " spark.read\r\n", " .format(\"com.microsoft.sqlserver.jdbc.spark\")\r\n", " .option(\"url\", url)\r\n", " .option(\"dbtable\", dbtable)\r\n", " .option(\"user\", user)\r\n", " .option(\"password\", password)\r\n", " .option(\"dataPoolDataSource\", datasource_name)\r\n", " .load()\r\n", " }" ], "metadata": { "azdata_cell_guid": "b083528c-8e6b-4c77-8b96-b60d043cbcbd" }, "outputs": [], "execution_count": 0 }, { "cell_type": "code", "source": [ "val new_df = df_read(dbtable, url, dataPoolDataSource=datasource_name)\r\n", "new_df.count" ], "metadata": { "azdata_cell_guid": "5f400a06-9b84-4885-b0e3-ecfdd63f8938" }, "outputs": [], "execution_count": 0 }, { "cell_type": "markdown", "source": [ "## **8. Stop the TwitterStream**" ], "metadata": { "azdata_cell_guid": "3d0bbecf-38e0-4883-84a3-107cfd556522" } }, { "cell_type": "code", "source": [ "twitterStream.stop()" ], "metadata": { "azdata_cell_guid": "7e995a68-2da1-4a52-8bd6-7ef04d9c8f49" }, "outputs": [], "execution_count": 0 } ] }