Initial Commit

Author: LearningJournal

09-SlidingWindowDemo/SlidingWindowDemo.py

@@ -1,55 +1,53 @@
 from pyspark.sql import SparkSession
-from pyspark.sql.functions import from_json, col, to_timestamp, window
+from pyspark.sql.functions import from_json, col, to_timestamp, window, max
 from pyspark.sql.types import StructType, StructField, StringType, DoubleType
 
 from lib.logger import Log4j
 
 if __name__ == "__main__":
     spark = SparkSession \
         .builder \
-        .appName("Sliding Window Demo") \
         .master("local[3]") \
+        .appName("Sliding Window Demo") \
         .config("spark.streaming.stopGracefullyOnShutdown", "true") \
+        .config("spark.sql.shuffle.partitions", 1) \
         .getOrCreate()
 
     logger = Log4j(spark)
 
     invoice_schema = StructType([
-        StructField("InvoiceNumber", StringType()),
         StructField("CreatedTime", StringType()),
-        StructField("StoreID", StringType()),
-        StructField("TotalAmount", DoubleType())
+        StructField("Reading", DoubleType())
     ])
 
-    kafka_df = spark.readStream \
+    kafka_source_df = spark \
+        .readStream \
         .format("kafka") \
         .option("kafka.bootstrap.servers", "localhost:9092") \
-        .option("subscribe", "invoices") \
+        .option("subscribe", "sensor") \
         .option("startingOffsets", "earliest") \
         .load()
 
-    value_df = kafka_df.select(from_json(col("value").cast("string"), invoice_schema).alias("value"))
-
-    # value_df.printSchema()
-    # value_df.show(truncate=False)
-
-    invoice_df = value_df.select("value.*") \
-        .withColumn("CreatedTime", to_timestamp("CreatedTime", "yyyy-MM-dd HH:mm:ss"))
+    value_df = kafka_source_df.select(col("key").cast("string").alias("SensorID"),
+                                      from_json(col("value").cast("string"), invoice_schema).alias("value"))
 
-    count_df = invoice_df.groupBy("StoreID",
-                                  window("CreatedTime", "5 minute", "1 minute")).count()
+    sensor_df = value_df.select("SensorID", "value.*") \
+        .withColumn("CreatedTime", to_timestamp(col("CreatedTime"), "yyyy-MM-dd HH:mm:ss"))
 
-    # count_df.printSchema()
-    # count_df.show(truncate=False)
+    agg_df = sensor_df \
+        .withWatermark("CreatedTime", "30 minute") \
+        .groupBy(col("SensorID"),
+                 window(col("CreatedTime"), "15 minute", "5 minute")) \
+        .agg(max("Reading").alias("MaxReading"))
 
-    output_df = count_df.select("StoreID", "window.start", "window.end", "count")
+    output_df = agg_df.select("SensorID", "window.start", "window.end", "MaxReading")
 
-    windowQuery = output_df.writeStream \
+    window_query = output_df.writeStream \
         .format("console") \
         .outputMode("update") \
         .option("checkpointLocation", "chk-point-dir") \
         .trigger(processingTime="1 minute") \
         .start()
 
-    logger.info("Counting Invoices")
-    windowQuery.awaitTermination()
+    logger.info("Waiting for Query")
+    window_query.awaitTermination()

09-SlidingWindowDemo/data/samples.json

@@ -1,3 +1,9 @@
-STR1534:{"InvoiceNumber": 101,"CreatedTime": "2019-02-05 10:01:00","StoreID": "STR1534", "TotalAmount": 1920}
-STR1534:{"InvoiceNumber": 103,"CreatedTime": "2019-02-05 10:03:19","StoreID": "STR1534", "TotalAmount": 2400}
-STR1534:{"InvoiceNumber": 105,"CreatedTime": "2019-02-05 10:07:50","StoreID": "STR1534", "TotalAmount": 6375}
+SET41:{"CreatedTime": "2019-02-05 09:54:00","Reading": 36.2}
+SET41:{"CreatedTime": "2019-02-05 09:59:00","Reading": 36.5}
+SET41:{"CreatedTime": "2019-02-05 10:04:00","Reading": 36.8}
+SET41:{"CreatedTime": "2019-02-05 10:09:00","Reading": 36.2}
+SET41:{"CreatedTime": "2019-02-05 10:14:00","Reading": 36.5}
+SET41:{"CreatedTime": "2019-02-05 10:19:00","Reading": 36.3}
+SET41:{"CreatedTime": "2019-02-05 10:24:00","Reading": 37.7}
+SET41:{"CreatedTime": "2019-02-05 10:29:00","Reading": 37.2}
+

09-SlidingWindowDemo/kafka-scripts/03-create-input-topic.cmd

@@ -1 +1 @@
-%KAFKA_HOME%\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic invoices
+%KAFKA_HOME%\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic sensor

09-SlidingWindowDemo/kafka-scripts/05-start-producer.cmd

@@ -1 +1 @@
-%KAFKA_HOME%\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic invoices --property "parse.key=true" --property "key.separator=:"
+%KAFKA_HOME%\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic sensor --property "parse.key=true" --property "key.separator=:"