Process Function

    • events (stream elements)
    • state (fault-tolerant, consistent, only on keyed stream)
    • timers (event time and processing time, only on keyed stream)

    The ProcessFunction can be thought of as a FlatMapFunction with access to keyed state and timers. It handles events by being invoked for each event received in the input stream(s).

    For fault-tolerant state, the ProcessFunction gives access to Flink’s keyed state, accessible via the RuntimeContext, similar to the way other stateful functions can access keyed state.

    The timers allow applications to react to changes in processing time and in . Every call to the function processElement(...) gets a Context object which gives access to the element’s event time timestamp, and to the TimerService. The TimerService can be used to register callbacks for future event-/processing-time instants. With event-time timers, the onTimer(...) method is called when the current watermark is advanced up to or beyond the timestamp of the timer, while with processing-time timers, onTimer(...) is called when wall clock time reaches the specified time. During that call, all states are again scoped to the key with which the timer was created, allowing timers to manipulate keyed state.

    To realize low-level operations on two inputs, applications can use CoProcessFunction or KeyedCoProcessFunction. This function is bound to two different inputs and gets individual calls to processElement1(...) and processElement2(...) for records from the two different inputs.

    Implementing a low level join typically follows this pattern:

    • Create a state object for one input (or both)
    • Update the state upon receiving elements from its input
    • Upon receiving elements from the other input, probe the state and produce the joined result

    For example, you might be joining customer data to financial trades, while keeping state for the customer data. If you care about having complete and deterministic joins in the face of out-of-order events, you can use a timer to evaluate and emit the join for a trade when the watermark for the customer data stream has passed the time of that trade.

    In the following example a KeyedProcessFunction maintains counts per key, and emits a key/count pair whenever a minute passes (in event time) without an update for that key:

    • The count, key, and last-modification-timestamp are stored in a ValueState, which is implicitly scoped by key.
    • For each record, the KeyedProcessFunction increments the counter and sets the last-modification timestamp
    • The function also schedules a callback one minute into the future (in event time)
    • Upon each callback, it checks the callback’s event time timestamp against the last-modification time of the stored count and emits the key/count if they match (i.e., no further update occurred during that minute)

    Java

    1. import org.apache.flink.api.common.state.ValueState;
    2. import org.apache.flink.api.common.state.ValueStateDescriptor;
    3. import org.apache.flink.api.java.tuple.Tuple;
    4. import org.apache.flink.api.java.tuple.Tuple2;
    5. import org.apache.flink.configuration.Configuration;
    6. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    7. import org.apache.flink.streaming.api.functions.KeyedProcessFunction.Context;
    8. import org.apache.flink.streaming.api.functions.KeyedProcessFunction.OnTimerContext;
    9. import org.apache.flink.util.Collector;
    12. // the source data stream
    13. DataStream<Tuple2<String, String>> stream = ...;
    15. // apply the process function onto a keyed stream
    16. DataStream<Tuple2<String, Long>> result = stream
    17.     .keyBy(value -> value.f0)
    18.     .process(new CountWithTimeoutFunction());
    20. /**
    21.  * The data type stored in the state
    22.  */
    23. public class CountWithTimestamp {
    25.     public String key;
    26.     public long count;
    27.     public long lastModified;
    28. }
    30. /**
    31.  * The implementation of the ProcessFunction that maintains the count and timeouts
    32.  */
    33. public class CountWithTimeoutFunction 
    34.         extends KeyedProcessFunction<Tuple, Tuple2<String, String>, Tuple2<String, Long>> {
    36.     /** The state that is maintained by this process function */
    37.     private ValueState<CountWithTimestamp> state;
    39.     @Override
    40.     public void open(Configuration parameters) throws Exception {
    41.         state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
    42.     }
    44.     @Override
    45.     public void processElement(
    46.             Tuple2<String, String> value, 
    47.             Context ctx, 
    48.             Collector<Tuple2<String, Long>> out) throws Exception {
    50.         // retrieve the current count
    51.         CountWithTimestamp current = state.value();
    52.         if (current == null) {
    53.             current = new CountWithTimestamp();
    54.             current.key = value.f0;
    55.         }
    57.         // update the state's count
    58.         current.count++;
    60.         // set the state's timestamp to the record's assigned event time timestamp
    61.         current.lastModified = ctx.timestamp();
    63.         // write the state back
    64.         state.update(current);
    66.         // schedule the next timer 60 seconds from the current event time
    67.         ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);
    68.     }
    70.     @Override
    71.     public void onTimer(
    72.             long timestamp, 
    73.             OnTimerContext ctx, 
    74.             Collector<Tuple2<String, Long>> out) throws Exception {
    75.         // get the state for the key that scheduled the timer
    76.         CountWithTimestamp result = state.value();
    78.         // check if this is an outdated timer or the latest timer
    79.         if (timestamp == result.lastModified + 60000) {
    80.             // emit the state on timeout
    81.             out.collect(new Tuple2<String, Long>(result.key, result.count));
    82.         }
    83.     }
    84. }

    Scala

    1. import org.apache.flink.api.common.state.ValueState
    2. import org.apache.flink.api.common.state.ValueStateDescriptor
    3. import org.apache.flink.api.java.tuple.Tuple
    4. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    5. import org.apache.flink.util.Collector
    7. // the source data stream
    8. val stream: DataStream[Tuple2[String, String]] = ...
    10. // apply the process function onto a keyed stream
    11. val result: DataStream[Tuple2[String, Long]] = stream
    12.   .keyBy(_._1)
    13.   .process(new CountWithTimeoutFunction())
    15. /**
    16.   * The data type stored in the state
    17.   */
    18. case class CountWithTimestamp(key: String, count: Long, lastModified: Long)
    20. /**
    21.   * The implementation of the ProcessFunction that maintains the count and timeouts
    22.   */
    23. class CountWithTimeoutFunction extends KeyedProcessFunction[Tuple, (String, String), (String, Long)] {
    25.   /** The state that is maintained by this process function */
    26.   lazy val state: ValueState[CountWithTimestamp] = getRuntimeContext
    27.     .getState(new ValueStateDescriptor[CountWithTimestamp]("myState", classOf[CountWithTimestamp]))
    30.   override def processElement(
    31.       value: (String, String), 
    32.       ctx: KeyedProcessFunction[Tuple, (String, String), (String, Long)]#Context, 
    33.       out: Collector[(String, Long)]): Unit = {
    35.     // initialize or retrieve/update the state
    36.     val current: CountWithTimestamp = state.value match {
    37.       case null =>
    38.         CountWithTimestamp(value._1, 1, ctx.timestamp)
    39.       case CountWithTimestamp(key, count, lastModified) =>
    40.         CountWithTimestamp(key, count + 1, ctx.timestamp)
    41.     }
    43.     // write the state back
    44.     state.update(current)
    46.     // schedule the next timer 60 seconds from the current event time
    47.     ctx.timerService.registerEventTimeTimer(current.lastModified + 60000)
    48.   }
    50.   override def onTimer(
    51.       timestamp: Long, 
    52.       ctx: KeyedProcessFunction[Tuple, (String, String), (String, Long)]#OnTimerContext, 
    53.       out: Collector[(String, Long)]): Unit = {
    55.     state.value match {
    56.       case CountWithTimestamp(key, count, lastModified) if (timestamp == lastModified + 60000) =>
    57.         out.collect((key, count))
    58.       case _ =>
    59.     }
    60.   }
    61. }

    Python

    1. import datetime
    3. from pyflink.common import Row, WatermarkStrategy
    4. from pyflink.common.typeinfo import Types
    5. from pyflink.common.watermark_strategy import TimestampAssigner
    6. from pyflink.datastream import StreamExecutionEnvironment
    7. from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
    8. from pyflink.datastream.state import ValueStateDescriptor
    9. from pyflink.table import StreamTableEnvironment
    12. class CountWithTimeoutFunction(KeyedProcessFunction):
    14.     def __init__(self):
    15.         self.state = None
    17.     def open(self, runtime_context: RuntimeContext):
    18.         self.state = runtime_context.get_state(ValueStateDescriptor(
    19.             "my_state", Types.PICKLED_BYTE_ARRAY()))
    21.         # retrieve the current count
    22.         current = self.state.value()
    23.         if current is None:
    24.             current = Row(value.f1, 0, 0)
    26.         # update the state's count
    29.         # set the state's timestamp to the record's assigned event time timestamp
    30.         current[2] = ctx.timestamp()
    32.         # write the state back
    33.         self.state.update(current)
    35.         # schedule the next timer 60 seconds from the current event time
    36.         ctx.timer_service().register_event_time_timer(current[2] + 60000)
    38.     def on_timer(self, timestamp: int, ctx: 'KeyedProcessFunction.OnTimerContext'):
    39.         # get the state for the key that scheduled the timer
    40.         result = self.state.value()
    42.         # check if this is an outdated timer or the latest timer
    43.         if timestamp == result[2] + 60000:
    44.             # emit the state on timeout
    45.             yield result[0], result[1]
    48. class MyTimestampAssigner(TimestampAssigner):
    50.     def __init__(self):
    51.         self.epoch = datetime.datetime.utcfromtimestamp(0)
    53.     def extract_timestamp(self, value, record_timestamp) -> int:
    54.         return int((value[0] - self.epoch).total_seconds() * 1000)
    57. if __name__ == '__main__':
    58.     env = StreamExecutionEnvironment.get_execution_environment()
    59.     t_env = StreamTableEnvironment.create(stream_execution_environment=env)
    61.     t_env.execute_sql("""
    62.             CREATE TABLE my_source (
    63.               a TIMESTAMP(3),
    64.               b VARCHAR,
    65.               c VARCHAR
    66.             ) WITH (
    67.               'connector' = 'datagen',
    68.               'rows-per-second' = '10'
    69.             )
    70.         """)
    72.     stream = t_env.to_append_stream(
    73.         t_env.from_path('my_source'),
    74.         Types.ROW([Types.SQL_TIMESTAMP(), Types.STRING(), Types.STRING()]))
    75.     watermarked_stream = stream.assign_timestamps_and_watermarks(
    76.         WatermarkStrategy.for_monotonous_timestamps()
    77.                          .with_timestamp_assigner(MyTimestampAssigner()))
    79.     # apply the process function onto a keyed stream
    80.     result = watermarked_stream.key_by(lambda value: value[1]) \
    81.                                .process(CountWithTimeoutFunction()) \
    82.                                .print()
    83.     env.execute()

    Java

    1. @Override
    2. public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {
    3.     K key = ctx.getCurrentKey();
    4.     // ...
    5. }

    Scala

    1. override def onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[OUT]): Unit = {
    2.   var key = ctx.getCurrentKey
    3.   // ...
    4. }

    Python

    Both types of timers (processing-time and event-time) are internally maintained by the TimerService and enqueued for execution.

    The TimerService deduplicates timers per key and timestamp, i.e., there is at most one timer per key and timestamp. If multiple timers are registered for the same timestamp, the onTimer() method will be called just once.

    Flink synchronizes invocations of onTimer() and processElement(). Hence, users do not have to worry about concurrent modification of state.

    Timers are fault tolerant and checkpointed along with the state of the application. In case of a failure recovery or when starting an application from a savepoint, the timers are restored.

    Timer Coalescing

    Since Flink maintains only one timer per key and timestamp, you can reduce the number of timers by reducing the timer resolution to coalesce them.

    For a timer resolution of 1 second (event or processing time), you can round down the target time to full seconds. Timers will fire at most 1 second earlier but not later than requested with millisecond accuracy. As a result, there are at most one timer per key and second.

    Java

    1. long coalescedTime = ((ctx.timestamp() + timeout) / 1000) * 1000;
    2. ctx.timerService().registerProcessingTimeTimer(coalescedTime);

    Scala

    1. val coalescedTime = ((ctx.timestamp + timeout) / 1000) * 1000
    2. ctx.timerService.registerProcessingTimeTimer(coalescedTime)
    1. coalesced_time = ((ctx.timestamp() + timeout) // 1000) * 1000
    2. ctx.timer_service().register_processing_time_timer(coalesced_time)

    Since event-time timers only fire with watermarks coming in, you may also schedule and coalesce these timers with the next watermark by using the current one:

    Java

    1. long coalescedTime = ctx.timerService().currentWatermark() + 1;
    2. ctx.timerService().registerEventTimeTimer(coalescedTime);

    Scala

    1. val coalescedTime = ctx.timerService.currentWatermark + 1
    2. ctx.timerService.registerEventTimeTimer(coalescedTime)

    Python

    Timers can also be stopped and removed as follows:

    Stopping a processing-time timer:

    Java

    1. long timestampOfTimerToStop = ...;
    2. ctx.timerService().deleteProcessingTimeTimer(timestampOfTimerToStop);

    Scala

    1. val timestampOfTimerToStop = ...
    2. ctx.timerService.deleteProcessingTimeTimer(timestampOfTimerToStop)

    Python

    1. timestamp_of_timer_to_stop = ...
    2. ctx.timer_service().delete_processing_time_timer(timestamp_of_timer_to_stop)

    Stopping an event-time timer:

    Java

    1. long timestampOfTimerToStop = ...;
    2. ctx.timerService().deleteEventTimeTimer(timestampOfTimerToStop);

    Scala

    1. val timestampOfTimerToStop = ...

    Python