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.

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

Wir verwenden Cookies, damit wir nachvollziehen können, wie Sie unsere Website verwenden, und um Ihr Erlebnis zu optimieren. Klicken Sie hier, wenn Sie mehr erfahren oder Ihre Cookie-Einstellungen ändern möchten. Wenn Sie weiter auf dieser Website surfen, stimmen Sie unserer Nutzung von Cookies zu.