sparkStreaming_basic_2.ipynb

{"cells":[{"cell_type":"code","source":["#https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html\n#https://databricks.com/blog/2017/02/23/working-complex-data-formats-structured-streaming-apache-spark-2-1.html\n#https://databricks.com/blog/2017/01/19/real-time-streaming-etl-structured-streaming-apache-spark-2-1.html\n\n#The key idea in Structured Streaming is to treat a live data stream as a table that is being continuously appended"],"metadata":{},"outputs":[],"execution_count":1},{"cell_type":"code","source":["#3. start writing stream to memory \n#dfNew.writeStream\\\n#.format(\"memory\")\\\n#.queryName(\"test\")\\\n#.outputMode(\"complete\")\\\n#.start()"],"metadata":{},"outputs":[],"execution_count":2},{"cell_type":"code","source":["#The “Output” is defined as what gets written out to the external storage.\n#Complete Mode - The entire updated Result Table will be written to the external storage. \n#Append Mode - Only the new rows appended in the Result Table since the last trigger will be written to the external storage.\n#Update Mode - Only the rows that were updated in the Result Table since the last trigger will be written to the external storage "],"metadata":{},"outputs":[],"execution_count":3},{"cell_type":"code","source":["#Handling Event-time and Late Data\n"],"metadata":{},"outputs":[],"execution_count":4},{"cell_type":"code","source":["#Input Sources\n# - File source, - Kafka source, - Socket source , - Rate source\n"],"metadata":{},"outputs":[],"execution_count":5},{"cell_type":"code","source":["#Schema inference and partition of streaming DataFrames/Datasets\n"],"metadata":{},"outputs":[],"execution_count":6},{"cell_type":"code","source":["socketDF = spark \\\n    .readStream \\\n    .format(\"socket\") \\\n    .option(\"host\", \"localhost\") \\\n    .option(\"port\", 9999) \\\n    .load()\n\nsocketDF.isStreaming    # Returns True for DataFrames that have streaming sources\n\nuserSchema = socketDF.schema\n\n# Read all the csv files written atomically in a directory\n#userSchema = StructType().add(\"name\", \"string\").add(\"age\", \"integer\")\ncsvDF = spark \\\n    .readStream \\\n    .option(\"sep\", \" \") \\\n    .schema(userSchema) \\\n    .csv(\"/FileStore/tables/\")\n  # Equivalent to format(\"csv\").load(\"/path/to/directory\")"],"metadata":{},"outputs":[],"execution_count":7},{"cell_type":"code","source":["runningStream = csvDF.writeStream\\\n.format(\"memory\")\\\n.queryName(\"myResult\")\\\n.outputMode(\"update\")\\\n.start()\n#ERROR\n#u'Complete output mode not supported when there are no streaming aggregations on streaming DataFrames/Datasets;;\\nFileSource[/FileStore/tables/]'.start()"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"code","source":["runningStream.stop()"],"metadata":{},"outputs":[],"execution_count":9},{"cell_type":"code","source":["#Schema inference and partition of streaming DataFrames/Datasets\n"],"metadata":{},"outputs":[],"execution_count":10},{"cell_type":"code","source":["#Window Operations on Event Time\nfrom pyspark.sql.functions import window\n#userSchema = StructType().add(\"name\", \"string\").add(\"age\", \"integer\")\n\n# grounBy aggregation in every 2 minutes window, sliding every 1 minute\n\nwindowedCount = csvDF.groupBy(window(csvDF[0], \"2 minutes\", \"1 minutes\"), csvDF[0] ).count()\n"],"metadata":{},"outputs":[],"execution_count":11},{"cell_type":"code","source":["runningStreamWindow = windowedCount.writeStream\\\n.format(\"memory\")\\\n.queryName(\"myResult\")\\\n.outputMode(\"complete\")\\\n.start()"],"metadata":{},"outputs":[],"execution_count":12},{"cell_type":"code","source":["runningStreamWindow.stop()"],"metadata":{},"outputs":[],"execution_count":13},{"cell_type":"code","source":["#Handling Late Data and Watermarking\n#Now consider what happens if one of the events arrives late to the application. For example, say, a word generated at 12:04 (i.e. event time) could be received by the application at 12:11. The application should use the time 12:04 instead of 12:11 to update the older counts for the window 12:00 - 12:10. \n#To enable this, in Spark 2.1, we have introduced watermarking, which lets the engine automatically track the current event time in the data and attempt to clean up old state accordingly.\n\nrunningStreamWindowWatermark = csvDF.withWatermark(\"timestamp\", \"5 minutes\")\\\n.groupBy(window(csvDF[0], \"2 minutes\",\"1 minutes\")).count()"],"metadata":{},"outputs":[],"execution_count":14}],"metadata":{"name":"sparkStreaming_basic_2","notebookId":3388231702265442},"nbformat":4,"nbformat_minor":0}