Project Metamorphosis: Wir präsentieren die Event-Streaming-Plattform der nächsten GenerationMehr Erfahren

Streaming-ETL

Berechnung von Bus-Verzögerungen

Many events happen on a time schedule where there is a scheduled time of arrival and an actual time of arrival. With these two data points, you are able to track the amount of time in which your event was delayed. By tracking this, you can better predict future delays and monitor your real-time progress.

In this example, we will use buses to demonstrate how we can calculate in real time the delays for every bus as it executes its route. After loading your static bus schedule into a KTable, you can then populate a live feed showing the number of minutes a bus is delayed from its expected course.

Visit the new Kafka-Tutorials site for the latest code examples

Directions

1. The bus schedule with expected arrival times is held as JSON data on a Kafka topic, with the ETA stored as a Unix timestamp. Register a KSQL table on this topic:

CREATE TABLE BUS_SCHEDULE (ROUTE_ID INTEGER, 
                           STOP_NUM INTEGER, 
                           ETA BIGINT) 
                     WITH (VALUE_FORMAT='JSON', 
                           KAFKA_TOPIC='bus-sched', 
                          KEY='STOP_NUM');

2. As buses complete their routes, they send events detailing the Unix timestamp of their arrival at a stop to a Kafka topic. Register this topic as a KSQL stream:

CREATE STREAM BUS_EVENTS (BUS_ID INTEGER, 
                          ROUTE_ID INTEGER, 
                          TIMESTAMP BIGINT, 
                          LAST_STOP INTEGER) 
                    WITH (VALUE_FORMAT='JSON', 
                          KAFKA_TOPIC='bus-events');

3. Configure KSQL to read all data from the beginning of a topic:

SET 'auto.offset.reset'='earliest';

4. Show a live feed of the number of minutes a bus is delayed from its expected course based on the route and stop number:

SELECT 'Bus: ' + CAST(EV.BUS_ID AS VARCHAR), 
       'Route: ' + CAST(EV.ROUTE_ID AS VARCHAR), 
       'Stop: ' + CAST(SCH.STOP_NUM AS VARCHAR), 
       'Due: ' + TIMESTAMPTOSTRING(SCH.ETA, 'yyyy-MM-dd  HH:mm:ss Z') AS ETA, 
       'Actual: ' + TIMESTAMPTOSTRING(EV.TIMESTAMP, 'HH:mm:ss Z') AS ACTUAL_TIME, 
       'Difference (minutes): ' + CAST((EV.TIMESTAMP-SCH.ETA)/1000/60 AS VARCHAR) AS MINUTES_DELAYED 
  FROM BUS_EVENTS EV 
       LEFT OUTER JOIN 
         BUS_SCHEDULE SCH 
         ON EV.LAST_STOP = SCH.STOP_NUM 
WHERE  EV.ROUTE_ID = SCH.ROUTE_ID;
Bus: 1 | Route: 1 | Stop: 0 | Due: 2019-01-01  10:00:00 +0000 | Actual: 10:00:00 +0000 | Difference (minutes): 0
Bus: 1 | Route: 1 | Stop: 1 | Due: 2019-01-01  10:10:00 +0000 | Actual: 10:12:00 +0000 | Difference (minutes): 2
Bus: 1 | Route: 1 | Stop: 2 | Due: 2019-01-01  10:25:00 +0000 | Actual: 10:28:00 +0000 | Difference (minutes): 3
< Back to the Stream Processing Cookbook