-
Notifications
You must be signed in to change notification settings - Fork 13
Expand file tree
/
Copy pathImplementing the speed layer of lambda architecture using Structured Spark Streaming
More file actions
1 lines (1 loc) · 29.2 KB
/
Copy pathImplementing the speed layer of lambda architecture using Structured Spark Streaming
File metadata and controls
1 lines (1 loc) · 29.2 KB
1
{"paragraphs":[{"text":"%md \n\n# Implementing the speed layer of lambda architecture using Structured Spark Streaming\n\n### Purpose: \n- provide analytics on real time data (\"intra day\") which batch layer cannot efficiently achieve\n- achieve this by:\n - ingest latest tweets from Kafka Producer and analtze only those for the current day \n - perform aggregations over the data to get the desired output of speed layer\n\n### Contents: \n- Configuring spark\n- Spark Structured Streaming\n - Input stage - defining the data source\n - Result stage - performing transformations on the stream\n - Output stage\n- Connecting to redshift cluster\n- Exporting data to Redshift","user":"anonymous","dateUpdated":"2017-11-11T15:55:51+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h1>Implementing the speed layer of lambda architecture using Structured Spark Streaming</h1>\n<h3>Purpose:</h3>\n<ul>\n <li>provide analytics on real time data (“intra day”) which batch layer cannot efficiently achieve</li>\n <li>achieve this by:\n <ul>\n <li>ingest latest tweets from Kafka Producer and analtze only those for the current day</li>\n <li>perform aggregations over the data to get the desired output of speed layer</li>\n </ul>\n </li>\n</ul>\n<h3>Contents:</h3>\n<ul>\n <li>Configuring spark</li>\n <li>Spark Structured Streaming\n <ul>\n <li>Input stage - defining the data source</li>\n <li>Result stage - performing transformations on the stream</li>\n <li>Output stage</li>\n </ul>\n </li>\n <li>Connecting to redshift cluster</li>\n <li>Exporting data to Redshift</li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509909863130_-1135889452","id":"20171105-202423_1160220584","dateCreated":"2017-11-05T20:24:23+0100","dateStarted":"2017-11-11T15:55:51+0100","dateFinished":"2017-11-11T15:55:51+0100","status":"FINISHED","progressUpdateIntervalMs":500,"focus":true,"$$hashKey":"object:273"},{"text":"%dep\n\nz.load(\"/Volumes/SD/Downloads/RedshiftJDBC42-1.2.10.1009.jar\")","user":"anonymous","dateUpdated":"2017-11-11T16:24:36+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"res0: org.apache.zeppelin.dep.Dependency = org.apache.zeppelin.dep.Dependency@7dfb0e5a\n"}]},"apps":[],"jobName":"paragraph_1509878460764_1962878483","id":"20171105-114100_1870675802","dateCreated":"2017-11-05T11:41:00+0100","dateStarted":"2017-11-11T16:24:36+0100","dateFinished":"2017-11-11T16:24:44+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:274"},{"text":"%md \n\n### Requirements","user":"anonymous","dateUpdated":"2017-11-11T14:30:05+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Requirements</h3>\n</div>"}]},"apps":[],"jobName":"paragraph_1510394086125_-1443536682","id":"20171111-105446_1758804734","dateCreated":"2017-11-11T10:54:46+0100","dateStarted":"2017-11-11T14:30:05+0100","dateFinished":"2017-11-11T14:30:05+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:275"},{"text":"import org.apache.spark.sql.SparkSession\nimport org.apache.spark.sql.streaming.ProcessingTime\nimport java.util.concurrent._","user":"anonymous","dateUpdated":"2017-11-11T16:24:59+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala","editOnDblClick":true},"editorMode":"ace/mode/scala","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"import org.apache.spark.sql.SparkSession\nimport org.apache.spark.sql.streaming.ProcessingTime\nimport java.util.concurrent._\n"}]},"apps":[],"jobName":"paragraph_1510394103383_-688043084","id":"20171111-105503_1480409678","dateCreated":"2017-11-11T10:55:03+0100","dateStarted":"2017-11-11T16:25:00+0100","dateFinished":"2017-11-11T16:25:18+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:276"},{"text":"%md \n\n### Configuring Spark\n- properly configuring spark for our workload\n- defining case class for tweets which will be used later on","user":"anonymous","dateUpdated":"2017-11-11T14:30:07+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Configuring Spark</h3>\n<ul>\n <li>properly configuring spark for our workload</li>\n <li>defining case class for tweets which will be used later on</li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509911366990_1694144500","id":"20171105-204926_42354742","dateCreated":"2017-11-05T20:49:26+0100","dateStarted":"2017-11-11T14:30:07+0100","dateFinished":"2017-11-11T14:30:07+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:277"},{"text":"Thread.sleep(5000)\n\nval spark = SparkSession\n .builder()\n .config(\"spark.sql.shuffle.partitions\",\"2\") // we are running this on my laptop\n .appName(\"Spark Structured Streaming example\")\n .getOrCreate()\n \ncase class tweet (id: String, created_at : String, followers_count: String, location : String, favorite_count : String, retweet_count : String)\n\nThread.sleep(5000)","user":"anonymous","dateUpdated":"2017-11-11T16:25:20+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"spark: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@20d06238\ndefined class tweet\n"}]},"apps":[],"jobName":"paragraph_1509779101105_500060833","id":"20171104-080501_1626533215","dateCreated":"2017-11-04T08:05:01+0100","dateStarted":"2017-11-11T16:25:20+0100","dateFinished":"2017-11-11T16:25:33+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:278"},{"text":"%md \n\n### Input stage - defining the data source\n- using Kafka as data source we specify:\n - location of kafka broker\n - relevant kafka topic\n - how to treat starting offsets","user":"anonymous","dateUpdated":"2017-11-11T14:30:09+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Input stage - defining the data source</h3>\n<ul>\n <li>using Kafka as data source we specify:\n <ul>\n <li>location of kafka broker</li>\n <li>relevant kafka topic</li>\n <li>how to treat starting offsets</li>\n </ul>\n </li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509881379573_-268499860","id":"20171105-122939_1561723250","dateCreated":"2017-11-05T12:29:39+0100","dateStarted":"2017-11-11T14:30:09+0100","dateFinished":"2017-11-11T14:30:09+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:279"},{"text":"var data_stream = spark\n .readStream // contantly expanding dataframe\n .format(\"kafka\")\n .option(\"kafka.bootstrap.servers\", \"localhost:9092\")\n .option(\"subscribe\", \"tweets-lambda1\")\n .option(\"startingOffsets\",\"latest\") //or latest\n .load()\n \n// note how similar API is to the batch version","user":"anonymous","dateUpdated":"2017-11-11T16:30:07+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"data_stream: org.apache.spark.sql.DataFrame = [key: binary, value: binary ... 5 more fields]\n"}]},"apps":[],"jobName":"paragraph_1509828449824_1485333558","id":"20171104-214729_426230385","dateCreated":"2017-11-04T21:47:29+0100","dateStarted":"2017-11-11T16:30:07+0100","dateFinished":"2017-11-11T16:30:14+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:280"},{"text":"data_stream.schema","user":"anonymous","dateUpdated":"2017-11-11T16:30:20+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"res8: org.apache.spark.sql.types.StructType = StructType(StructField(key,BinaryType,true), StructField(value,BinaryType,true), StructField(topic,StringType,true), StructField(partition,IntegerType,true), StructField(offset,LongType,true), StructField(timestamp,TimestampType,true), StructField(timestampType,IntegerType,true))\n"}]},"apps":[],"jobName":"paragraph_1509896872792_453692520","id":"20171105-164752_719330437","dateCreated":"2017-11-05T16:47:52+0100","dateStarted":"2017-11-11T16:30:20+0100","dateFinished":"2017-11-11T16:30:23+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:281"},{"text":"%md \n\n### Result stage - performing transformations on the stream\n- extract the value column of kafka message\n- parse each row into a member of tweet class\n- filter to only look at todays tweets as results\n- perform aggregations","user":"anonymous","dateUpdated":"2017-11-11T14:30:12+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Result stage - performing transformations on the stream</h3>\n<ul>\n <li>extract the value column of kafka message</li>\n <li>parse each row into a member of tweet class</li>\n <li>filter to only look at todays tweets as results</li>\n <li>perform aggregations</li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509912237119_-1371939219","id":"20171105-210357_553784609","dateCreated":"2017-11-05T21:03:57+0100","dateStarted":"2017-11-11T14:30:12+0100","dateFinished":"2017-11-11T14:30:12+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:282"},{"text":"var data_stream_cleaned = data_stream\n .selectExpr(\"CAST(value AS STRING) as string_value\")\n .as[String]\n .map(x => (x.split(\";\"))) //wrapped array\n .map(x => tweet(x(0), x(1), x(2), x(3), x(4), x(5)))\n .selectExpr( \"cast(id as long) id\", \"CAST(created_at as timestamp) created_at\", \"cast(followers_count as int) followers_count\", \"location\", \"cast(favorite_count as int) favorite_count\", \"cast(retweet_count as int) retweet_count\")\n .toDF()\n .filter(col(\"created_at\").gt(current_date())) // kafka will retain data for last 24 hours, this is needed because we are using complete mode as output\n .groupBy(\"location\")\n .agg(count(\"id\"), sum(\"followers_count\"), sum(\"favorite_count\"), sum(\"retweet_count\"))","user":"anonymous","dateUpdated":"2017-11-11T16:32:33+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"data_stream_cleaned: org.apache.spark.sql.DataFrame = [location: string, count(id): bigint ... 3 more fields]\n"}]},"apps":[],"jobName":"paragraph_1509838694466_-674508628","id":"20171105-003814_85265595","dateCreated":"2017-11-05T00:38:14+0100","dateStarted":"2017-11-11T16:32:33+0100","dateFinished":"2017-11-11T16:32:43+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:283"},{"text":"data_stream_cleaned.schema","user":"anonymous","dateUpdated":"2017-11-11T16:32:46+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"res9: org.apache.spark.sql.types.StructType = StructType(StructField(location,StringType,true), StructField(count(id),LongType,false), StructField(sum(followers_count),LongType,true), StructField(sum(favorite_count),LongType,true), StructField(sum(retweet_count),LongType,true))\n"}]},"apps":[],"jobName":"paragraph_1509896863262_62756989","id":"20171105-164743_19706666","dateCreated":"2017-11-05T16:47:43+0100","dateStarted":"2017-11-11T16:32:46+0100","dateFinished":"2017-11-11T16:32:47+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:284"},{"text":"%md \n\n### Output stage\n- specify the following:\n - data sink - exporting to memory (table can be accessed similar to registerTempTable()/ createOrReplaceTempView() function )\n - trigger - time between running the pipeline (ie. when to do: polling for new data, data transformation)\n - output mode - complete, append or update - since in Result stage we use aggregates, we can only use Complete or Update out put mode","user":"anonymous","dateUpdated":"2017-11-11T16:33:38+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Output stage</h3>\n<ul>\n <li>specify the following:\n <ul>\n <li>data sink - exporting to memory (table can be accessed similar to registerTempTable()/ createOrReplaceTempView() function )</li>\n <li>trigger - time between running the pipeline (ie. when to do: polling for new data, data transformation)</li>\n <li>output mode - complete, append or update - since in Result stage we use aggregates, we can only use Complete or Update out put mode</li>\n </ul>\n </li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509912259110_1463706147","id":"20171105-210419_1857413462","dateCreated":"2017-11-05T21:04:19+0100","dateStarted":"2017-11-11T16:33:38+0100","dateFinished":"2017-11-11T16:33:42+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:285"},{"text":"val query = data_stream_cleaned.writeStream\n .format(\"memory\")\n .queryName(\"demo\")\n .trigger(ProcessingTime(\"30 seconds\")) // means that that spark will look for new data only every minute\n .outputMode(\"complete\") // could also be append or update\n .start()","user":"anonymous","dateUpdated":"2017-11-11T16:34:18+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"query: org.apache.spark.sql.streaming.StreamingQuery = Streaming Query demo [id = 824a35c8-af8d-4b57-83c4-4cbf3c1f9418, runId = 5dcdee01-de96-403e-997e-350bfd5d8f2c] [state = ACTIVE]\n"}]},"apps":[],"jobName":"paragraph_1509873755450_-691017020","id":"20171105-102235_1294551369","dateCreated":"2017-11-05T10:22:35+0100","dateStarted":"2017-11-11T16:34:18+0100","dateFinished":"2017-11-11T16:34:22+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:286"},{"text":"query.status","user":"anonymous","dateUpdated":"2017-11-11T16:34:25+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"res10: org.apache.spark.sql.streaming.StreamingQueryStatus =\n{\n \"message\" : \"Processing new data\",\n \"isDataAvailable\" : true,\n \"isTriggerActive\" : true\n}\n"}]},"apps":[],"jobName":"paragraph_1510413138022_529774344","id":"20171111-161218_1171458195","dateCreated":"2017-11-11T16:12:18+0100","dateStarted":"2017-11-11T16:34:25+0100","dateFinished":"2017-11-11T16:34:26+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:287"},{"text":"query.explain","user":"anonymous","dateUpdated":"2017-11-11T16:34:35+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"== Physical Plan ==\n*HashAggregate(keys=[location#36], functions=[count(id#40L), sum(cast(followers_count#42 as bigint)), sum(cast(favorite_count#43 as bigint)), sum(cast(retweet_count#44 as bigint))])\n+- Exchange hashpartitioning(location#36, 2)\n +- *HashAggregate(keys=[location#36], functions=[partial_count(id#40L), partial_sum(cast(followers_count#42 as bigint)), partial_sum(cast(favorite_count#43 as bigint)), partial_sum(cast(retweet_count#44 as bigint))])\n +- *Project [cast(cast(id#33 as decimal(20,0)) as bigint) AS id#40L, cast(cast(followers_count#35 as decimal(20,0)) as int) AS followers_count#42, location#36, cast(cast(favorite_count#37 as decimal(20,0)) as int) AS favorite_count#43, cast(cast(retweet_count#38 as decimal(20,0)) as int) AS retweet_count#44]\n +- *Filter (isnotnull(created_at#34) && (cast(cast(created_at#34 as timestamp) as string) > cast(current_batch_timestamp(1510414470012, DateType) as string)))\n +- *SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).id, true) AS id#33, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).created_at, true) AS created_at#34, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).followers_count, true) AS followers_count#35, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).location, true) AS location#36, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).favorite_count, true) AS favorite_count#37, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, $line155526911422.$read$$iw$$iw$tweet, true], top level Product input object).retweet_count, true) AS retweet_count#38]\n +- *MapElements <function1>, obj#32: $line155526911422.$read$$iw$$iw$tweet\n +- *MapElements <function1>, obj#22: [Ljava.lang.String;\n +- *DeserializeToObject string_value#15.toString, obj#21: java.lang.String\n +- *Project [cast(value#141 as string) AS string_value#15]\n +- Scan ExistingRDD[key#140,value#141,topic#142,partition#143,offset#144L,timestamp#145,timestampType#146]\n"}]},"apps":[],"jobName":"paragraph_1510413000978_-581196590","id":"20171111-161000_1529325108","dateCreated":"2017-11-11T16:10:00+0100","dateStarted":"2017-11-11T16:34:35+0100","dateFinished":"2017-11-11T16:34:36+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:288"},{"text":"%spark.sql \n\nselect * from demo","user":"anonymous","dateUpdated":"2017-11-11T16:39:15+0100","config":{"colWidth":12,"enabled":true,"results":{"0":{"graph":{"mode":"table","height":300,"optionOpen":false},"helium":{}}},"editorSetting":{"language":"sql"},"editorMode":"ace/mode/sql"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TABLE","data":"location\tcount(id)\tsum(followers_count)\tsum(favorite_count)\tsum(retweet_count)\nYokohama\t2\t1110\t0\t115\nAngola, Luanda \t3\t1572\t0\t94221\nハッピーワールド\t4\t384\t0\t0\n推しが定まりません。そらワンいって考えます\t4\t6932\t0\t0\nTallinn, Estonia\t4\t248\t0\t464\n\t233\t310223\t0\t1344636\nOhio\t2\t28806\t0\t0\n京都市右京区\t3\t4272\t0\t0\n孤独\t3\t1059\t0\t0\nGeorgia, USA\t4\t3448\t0\t0\nManaus / São Paulo\t4\t1056\t0\t0\n当てます。\t3\t1299\t0\t0\nTriplex\t4\t7780\t0\t88\nPhilly✈️Tuskegee👉🏾D.C.\t3\t4842\t0\t3\nnew jersey\t4\t864\t0\t0\nさいたま市\t4\t3384\t0\t0\n日本 東京\t3\t2337\t0\t0\nD.O. 도경수\t3\t1428\t0\t3018\n千葉 船橋市\t4\t156\t0\t0\nEntroncamento, Portugal\t4\t1392\t0\t2473\nAntwerpen, Belgium\t3\t1629\t0\t0\nチ ン パ ン ．🌷 ￤ﾍｯﾀﾞｰ ⇒ ﾌﾐ !!!\t3\t669\t0\t208\nEau Claire, Wisconsin\t3\t254721\t0\t1\n@無所属 お仕事依頼DMへ📩\t3\t2406\t0\t0\nJeje l'omo Eko nlo\t3\t1008\t0\t0\nstaten island, new york\t3\t348\t0\t0\nlagos\t3\t3450\t0\t234\nAtlanta GA Metro\t3\t273\t0\t0\nBerks. UK\t3\t13383\t0\t66\nΑνδραβιδα\t3\t7398\t0\t5172\nFinland, no polar bears here\t3\t609\t0\t0\nGuayaquil, Ecuador\t3\t1827\t0\t120\n내가 하는 말이 있다면 그건 거짓말\t4\t20\t0\t0\nLouisville, KY\t3\t2250\t0\t0\nottawa\t4\t1024\t0\t0\nً\t4\t11140\t0\t122960\ncairo\t2\t706\t0\t0\nSan Francisco, CA\t3\t1167\t0\t0\n🇧🇸 // 🇯🇲\t3\t3522\t0\t111\nNorwalk, CA\t3\t765\t0\t0\n京都 - 大阪\t4\t2968\t0\t341\nBeen all around the world\t3\t1161\t0\t43983\nUSA\t6\t31782\t0\t0\n호그와트\t3\t183\t0\t1077\nSeabrook, TX\t3\t1023\t0\t9860\nNein. \t3\t5421\t0\t0\nかるわかのストレス清掃員♪\t3\t372\t0\t0\nRochester/London, UK\t3\t7557\t0\t12\nbuenos aires argentina\t3\t444\t0\t0\nHoseok's heart~\t4\t7268\t0\t0\nNorth West, England\t3\t909\t0\t1431\nGloucester, UK\t3\t5424\t0\t0\n愛知県岡崎市\t4\t1964\t0\t0\nアップルパイの中\t3\t759\t0\t0\nMakati City\t3\t3\t0\t129771\nCorrientes\t3\t411\t0\t1521\nYら,猫好き,車好き,ゲーム好きさんと繋がりたい🗿\t3\t630\t0\t0\n{May - Sofi - Dians - Paula}\t3\t48723\t4\t1\nToronto\t4\t3744\t0\t0\nどすこいパレード\t4\t1044\t0\t98\nMalaysia\t4\t1264\t0\t48736\nJapan Tokyo\t4\t32520\t0\t0\n#Adelaido 👑\t2\t2094\t6\t2\nJapan\t3\t126603\t0\t0\nPilipínas\t3\t1362\t0\t20979\nManaus, Brasil\t3\t3222\t0\t0\n127.0.0.1\t3\t5997\t0\t4143\nFigueres\t3\t4485\t0\t0\n運動会する\t4\t256\t0\t0\nRepublic of the Philippines\t3\t1347\t0\t54\nアバリスの膝の上\t3\t630\t0\t0\nPortugal\t2\t1476\t0\t1241\nOhio, USA\t4\t1332\t0\t0\nlincoln, england\t4\t2844\t0\t4\nCalifornia, USA\t3\t1281\t0\t402\nLondon, Ontario\t3\t7467\t0\t0\nOrlando, FL\t4\t1056\t0\t0\nLos Angeles, CA\t7\t29565\t0\t775\nAtlanta, GA\t3\t279\t0\t318\nmismuertos city\t3\t1449\t0\t57932\nカルデアの冷蔵庫\t3\t141\t0\t0\nBarueri, Brasil\t3\t2607\t0\t3\nahead\t3\t831\t1\t1\nparadise with drake\t3\t3426\t0\t0\nmiami\t3\t504\t0\t0\n栃木県\t4\t316\t0\t0\nJHB/SF\t4\t11420\t0\t0\n京都府 京都市 \t3\t18\t0\t0\nMichigan \t4\t3272\t0\t0\nAvellaneda, Buenos Aires\t3\t774\t0\t10923\n대한민국\t3\t63\t0\t1068\n日本\t19\t6294\t0\t231\nether \t4\t88\t0\t16\nPunjab, Pakistan\t2\t2088\t0\t0\nUnited Kingdom\t4\t1740\t0\t0\nToday's Market Movement\t3\t586\t0\t9\nOn Melancholy Hill\t3\t498\t0\t0\nLos Angeles\t4\t612\t0\t0\nMadrid, Comunidad de Madrid\t3\t7578\t0\t375\nMakati City, National Capital \t3\t660\t0\t36\nSingapore \t3\t588\t0\t204\nIndia\t3\t342\t0\t0\nNorth Riding/Samrand\t3\t315\t0\t29427\nBinghamton University\t3\t2166\t0\t0\nGurgaon, India\t4\t18184\t0\t0\n広島県\t3\t642\t0\t206\nみんなの心の中☆\t2\t204\t0\t0\n"}]},"apps":[],"jobName":"paragraph_1509873913879_-781586253","id":"20171105-102513_1944185364","dateCreated":"2017-11-05T10:25:13+0100","dateStarted":"2017-11-11T16:39:16+0100","dateFinished":"2017-11-11T16:39:16+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:289"},{"text":"%md\n\n### Connecting to redshift cluster\n- defining JDBC connection to connect to redshift","user":"anonymous","dateUpdated":"2017-11-11T14:30:16+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Connecting to redshift cluster</h3>\n<ul>\n <li>defining JDBC connection to connect to redshift</li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509909816175_369050675","id":"20171105-202336_1741159108","dateCreated":"2017-11-05T20:23:36+0100","dateStarted":"2017-11-11T14:30:16+0100","dateFinished":"2017-11-11T14:30:16+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:290"},{"text":"//create properties object\nClass.forName(\"com.amazon.redshift.jdbc42.Driver\")\n\nval prop = new java.util.Properties\nprop.setProperty(\"driver\", \"com.amazon.redshift.jdbc42.Driver\")\nprop.setProperty(\"user\", \"dorian\")\nprop.setProperty(\"password\", \"Demo1234\") \n\n//jdbc mysql url - destination database is named \"data\"\nval url = \"jdbc:redshift://data-warehouse.c3glymsgdgty.us-east-1.redshift.amazonaws.com:5439/lambda\"\n\n//destination database table \nval table = \"speed_layer\"","user":"anonymous","dateUpdated":"2017-11-11T16:36:02+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"res13: Class[_] = class com.amazon.redshift.jdbc42.Driver\nprop: java.util.Properties = {}\nres15: Object = null\nres16: Object = null\nres17: Object = null\nurl: String = jdbc:redshift://data-warehouse.c3glymsgdgty.us-east-1.redshift.amazonaws.com:5439/lambda\ntable: String = speed_layer\n"}]},"apps":[],"jobName":"paragraph_1509874084488_-1803262626","id":"20171105-102804_443283701","dateCreated":"2017-11-05T10:28:04+0100","dateStarted":"2017-11-11T16:36:02+0100","dateFinished":"2017-11-11T16:36:07+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:291"},{"text":"%md \n\n### Exporting data to Redshift\n- \"overwriting\" the table with results of query stored in memory as result of the speed layer\n- scheduling the function to run every hour\n","user":"anonymous","dateUpdated":"2017-11-11T14:30:18+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"markdown","editOnDblClick":true},"editorMode":"ace/mode/markdown","editorHide":true,"tableHide":false},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"HTML","data":"<div class=\"markdown-body\">\n<h3>Exporting data to Redshift</h3>\n<ul>\n <li>“overwriting” the table with results of query stored in memory as result of the speed layer</li>\n <li>scheduling the function to run every hour</li>\n</ul>\n</div>"}]},"apps":[],"jobName":"paragraph_1509881434795_1558195972","id":"20171105-123034_774658214","dateCreated":"2017-11-05T12:30:34+0100","dateStarted":"2017-11-11T14:30:18+0100","dateFinished":"2017-11-11T14:30:18+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:292"},{"text":"val df = spark.sql(\"select * from demo\")\n\n//write data from spark dataframe to database\ndf.write.mode(\"overwrite\").jdbc(url, table, prop)\n","user":"anonymous","dateUpdated":"2017-11-11T16:39:19+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"results":{"code":"SUCCESS","msg":[{"type":"TEXT","data":"df: org.apache.spark.sql.DataFrame = [location: string, count(id): bigint ... 3 more fields]\n"}]},"apps":[],"jobName":"paragraph_1509878690443_-615786910","id":"20171105-114450_965590151","dateCreated":"2017-11-05T11:44:50+0100","dateStarted":"2017-11-11T16:39:19+0100","dateFinished":"2017-11-11T16:39:30+0100","status":"FINISHED","progressUpdateIntervalMs":500,"$$hashKey":"object:293"},{"text":"","user":"anonymous","dateUpdated":"2017-11-11T16:37:05+0100","config":{"colWidth":12,"enabled":true,"results":{},"editorSetting":{"language":"scala"},"editorMode":"ace/mode/scala"},"settings":{"params":{},"forms":{}},"apps":[],"jobName":"paragraph_1509949807769_1425690320","id":"20171106-073007_1469584911","dateCreated":"2017-11-06T07:30:07+0100","status":"READY","errorMessage":"","progressUpdateIntervalMs":500,"$$hashKey":"object:294"}],"name":"Implementing the speed layer of lambda architecture using Structured Spark Streaming","id":"2CY6MSDVK","angularObjects":{"2CZTSJM9A:shared_process":[],"2CXD9FT1P:shared_process":[],"2CYBUJCZE:shared_process":[],"2CZXYWRND:shared_process":[],"2CWX9E9KA:shared_process":[],"2CWJS6R2N:shared_process":[],"2CW6U7X7Z:shared_process":[],"2CY9X3W1T:shared_process":[],"2CX93H291:shared_process":[],"2CZRYW3SZ:shared_process":[],"2CYT9Z9RC:shared_process":[],"2CY2R49R6:shared_process":[],"2CYQW36AU:shared_process":[],"2CWPRKMXH:shared_process":[],"2CWU95D3A:shared_process":[],"2CXJ7UBRF:shared_process":[],"2CWWTMY7M:shared_process":[],"2CY3EBJAE:shared_process":[],"2CYFQZER9:shared_process":[]},"config":{"looknfeel":"default","personalizedMode":"false"},"info":{}}