Lab: `ProcessFunction` and Timers (Long Ride Alerts)

The goal of the “Long Ride Alerts” exercise is to provide a warning whenever a taxi ride lasts for more than two hours.

This should be done using the event time timestamps and watermarks that are provided in the data stream.

The stream is out-of-order, and it is possible that the END event for a ride will be processed before its START event.

An END event may be missing, but you may assume there are no duplicated events, and no missing START events.

It is not enough to simply wait for the END event and calculate the duration, as we want to be alerted about the long ride as soon as possible.

You should eventually clear any state you create.

Input Data

The input data of this exercise is a DataStream of taxi ride events.

Expected Output

The result of the exercise should be a DataStream<LONG> that contains the rideId for rides with a duration that exceeds two hours.

The resulting stream should be printed to standard out.

Getting Started

:information_source: Rather than following these links to the sources, you might prefer to open these classes in your IDE.

Exercise Classes

Unit Tests

Integration Tests

Implementation Hints

This exercise revolves around using a KeyedProcessFunction to manage some state and event time timers, and doing so in a way that works even when the END event for a given rideId arrives before the START. The challenge is figuring out what state and timers to use, and when to set and clear the state (and timers).

Documentation

After you've completed the exercise

Read the discussion of the reference solutions.

Reference Solutions

Reference solutions:

Back to Labs Overview

Lab: ProcessFunction and Timers (Long Ride Alerts)