Merge branch 'master' of github.com:confluentinc/demo-scene

industry-themes/next_best_offer_banking/producers/producer_script.sh

@@ -1,8 +1,8 @@
-cat customers_fin.json | kafkacat -b localhost:9092 -t CUSTOMERS_STREAM
-cat offers.json | kafkacat -b localhost:9092 -t OFFERS_STREAM
+cat ../data/customers_fin.json | kafkacat -b localhost:9092 -t CUSTOMERS_STREAM
+cat ../data/offers.json | kafkacat -b localhost:9092 -t OFFERS_STREAM
 kafka-producer-perf-test \
     --topic CUSTOMER_ACTIVITY_STREAM \
     --throughput 1 \
     --producer-props bootstrap.servers=localhost:9092 \
-    --payload-file customer_activity.json \
-    --num-records 100000
+    --payload-file ../data/customer_activity.json \
+    --num-records 100000

industry-themes/next_best_offer_insurance/producers/producer_script.sh

@@ -1,8 +1,8 @@
-cat customers_insurance.json | kafkacat -b localhost:9092 -t CUSTOMERS_STREAM
-cat offers_insurance.json | kafkacat -b localhost:9092 -t OFFERS_STREAM
+cat ../data/customers_insurance.json | kafkacat -b localhost:9092 -t CUSTOMERS_STREAM
+cat ../data/offers_insurance.json | kafkacat -b localhost:9092 -t OFFERS_STREAM
 kafka-producer-perf-test \
     --topic CUSTOMER_ACTIVITY_STREAM \
     --throughput 1 \
     --producer-props bootstrap.servers=localhost:9092 \
-    --payload-file customer_activity_insurance.json \
-    --num-records 100000
+    --payload-file ../data/customer_activity_insurance.json \
+    --num-records 100000

industry-themes/truck_sensors/data/truck_ids.csv

@@ -0,0 +1,11 @@
+truck_id
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10

industry-themes/truck_sensors/ksql/truck_alerts.ksql

@@ -0,0 +1,90 @@
+SET 'auto.offset.reset'='earliest';
+
+CREATE STREAM TRUCK_ENGINE_SENSORS
+(
+    TRUCK_ID STRING,
+    ENGINE_TEMPERATURE INTEGER,
+    AVERAGE_RPM INTEGER
+)
+WITH (
+    KAFKA_TOPIC = 'TRUCK_ENGINE_SENSORS',
+    VALUE_FORMAT = 'JSON'
+);
+
+CREATE STREAM TRUCK_ENGINE_SENSORS_KEYED
+WITH (
+    KAFKA_TOPIC = 'TRUCK_ENGINE_SENSORS_KEYED',
+    VALUE_FORMAT = 'AVRO'
+) AS
+SELECT
+    ROWTIME AS EVENT_TIME,
+    TRUCK_ID,
+    ENGINE_TEMPERATURE,
+    AVERAGE_RPM
+FROM TRUCK_ENGINE_SENSORS
+WHERE TRUCK_ID NOT LIKE 'truck_id'
+PARTITION BY TRUCK_ID;
+
+CREATE STREAM TRUCK_LOCATION
+(
+    TRUCK_ID STRING,
+    TRUCK_LAT STRING,
+    TRUCK_LONG STRING
+)
+WITH (
+    KAFKA_TOPIC = 'TRUCK_LOCATION',
+    VALUE_FORMAT = 'JSON'
+);
+
+CREATE STREAM TRUCK_LOCATION_KEYED
+WITH (
+    KAFKA_TOPIC = 'TRUCK_LOCATION_KEYED',
+    VALUE_FORMAT = 'AVRO'
+) AS
+SELECT
+    ROWTIME AS EVENT_TIME,
+    TRUCK_ID,
+    TRUCK_LAT,
+    TRUCK_LONG
+FROM TRUCK_LOCATION
+PARTITION BY TRUCK_ID;
+
+CREATE STREAM TRUCK_EVENTS_JOINED
+WITH (
+    KAFKA_TOPIC = 'TRUCK_EVENTS_JOINED',
+    VALUE_FORMAT = 'AVRO'
+) AS
+SELECT
+    tl.EVENT_TIME AS LOCATION_EVENT_TIME,
+    tl.TRUCK_ID AS TRUCK_ID_LOC,
+    tl.TRUCK_LAT,
+    tl.TRUCK_LONG,
+    te.TRUCK_ID AS TRUCK_ID_ENG,
+    te.EVENT_TIME AS ENGINE_EVENT_TIME,
+    te.ENGINE_TEMPERATURE,
+    te.AVERAGE_RPM,
+    CASE
+        WHEN ENGINE_TEMPERATURE > 200 AND AVERAGE_RPM > 2000 THEN 1
+        ELSE 0
+    END AS HIGH_TEMP
+FROM TRUCK_LOCATION_KEYED tl
+LEFT OUTER JOIN TRUCK_ENGINE_SENSORS_KEYED te
+WITHIN 10 SECONDS
+ON tl.TRUCK_ID = te.TRUCK_ID;
+
+CREATE TABLE TRUCK_ALERTS
+WITH (
+    KAFKA_TOPIC = 'TRUCK_ALERTS',
+    VALUE_FORMAT = 'AVRO'
+) AS
+SELECT
+    TRUCK_ID_LOC,
+    COLLECT_SET('{' + TRUCK_LAT + ',' + TRUCK_LONG + '}') AS TRUCK_LAT_LONG_ARRAY,
+    MAX(ENGINE_TEMPERATURE) AS MAX_ENG_TEMP,
+    MAX(AVERAGE_RPM) AS MAX_AVG_RPM,
+    COUNT(*)
+FROM TRUCK_EVENTS_JOINED
+WINDOW TUMBLING (SIZE 4 SECONDS)
+WHERE HIGH_TEMP = 1
+GROUP BY TRUCK_ID_LOC
+HAVING COUNT(*) > 20;

industry-themes/truck_sensors/producers/producers.sh

@@ -0,0 +1,12 @@
+kafka-producer-perf-test \
+    --topic TRUCK_ENGINE_SENSORS \
+    --throughput 5 \
+    --producer-props bootstrap.servers=localhost:9092 \
+    --payload-file ../data/truck_engine_sensors.json \
+    --num-records 100000 &
+kafka-producer-perf-test \
+    --topic TRUCK_LOCATION \
+    --throughput 5 \
+    --producer-props bootstrap.servers=localhost:9092 \
+    --payload-file ../data/truck_location.json \
+    --num-records 100000 &