Assigner with periodic watermarks

Assigning watermarks periodically means that we instruct the system to check the progress of event time in fixed intervals of machine time. The default interval is set to 200 milliseconds but it can be configured using the ExecutionConfig.setAutoWatermarkInterval() method as shown in  Example 6-3.

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// generate watermarks every 5 seconds
env.getConfig.setAutoWatermarkInterval(5000)

In the above example, you instruct the program to check the current watermark value every 5 seconds. What actually happens is that every 5 seconds Flink invokes the getCurrentWatermark() method of AssignerWithPeriodicWatermarks. If the method returns a non-null value with a timestamp larger than the timestamp of the previous watermark, then the new watermark is forwarded. Note that this check is necessary to ensure that event time continuously increases. Otherwise, if the method returns a null value or the timestamp of the returned watermark is smaller than that of the last emitted one, no watermark is produced.

Example 6-4 shows an assigner with periodic timestamps which produces watermarks by keeping track of the maximum element timestamp it has seen so far. When being asked for a new watermark, the assigner returns a watermark with the maximum timestamp minus a 1 minute tolerance interval.

class PeriodicAssigner 
    extends AssignerWithPeriodicWatermarks[SensorReading] {

  val bound: Long = 60 * 1000     // 1 min in ms
  var maxTs: Long = Long.MinValue // the maximum observed timestamp

  override def getCurrentWatermark: Watermark = {
    // generated watermark with 1 min tolerance
    new Watermark(maxTs - bound)
  }

  override def extractTimestamp(
      r: SensorReading, 
      previousTS: Long): Long = {
    // update maximum timestamp
    maxTs = maxTs.max(r.timestamp)
    // return record timestamp
    r.timestamp
  }
}

The DataStream API provides implementations for two common cases of timestamp assigners with periodic watermarks. If your input elements have timestamps which are monotonously increasing, you can use the shortcut method assignAscendingTimestamps. This method uses the current timestamp to generate watermarks, since no earlier timestamps can appear. Example 6-5 shows how to generate watermarks for ascending timestamps.

val stream: DataStream[MyEvent] = ...
val withTimestampsAndWatermarks = stream
      .assignAscendingTimestamps(e => e.getCreationTime)

The other common case of periodic watermark generation is when you know the maximum lateness that you will encounter in the input stream, that is the maximum difference between an element’s timestamp and the largest timestamp of all perviously ingested elements. For such cases, Flink provides the BoundedOutOfOrdernessTimestampExtractor which takes the maximum expected lateness as an argument.

val stream: DataStream[MyEvent] = ...
val output = stream.assignTimestampsAndWatermarks(
 new BoundedOutOfOrdernessTimestampExtractor[MyEvent](
   Time.seconds(10))( _.getCreationTime)

In Example 6-6 elements are allowed to be late for 10 seconds. That is, if the difference between an element’s event time and the maximum timestamp of all previous elements is greater than 10 seconds, the element might arrive after a related computation has completed and result has been emitted. Flink offers different strategies to handle such late events and we discuss those later in this chapter.

Assigner with punctuated watermarks

Sometimes the input stream contains special tuples or markers that indicate the stream’s progress. For such cases or when watermarks can be defined based on some other property of the input elements, Flink provides the AssignerWithPunctuatedWatermarks. The interface contains the checkAndGetNextWatermark() method which is called for each event right after extractTimestamp(). The method can decide to generate a new watermark or not. A new watermark is emitted if the method returns a non-null watermark which is larger than the latest emitted watermark.

Example 6-7 shows a punctuated watermark assigner that emits a watermark for every reading that it rececives from the sensor with the id "sensor_1".

class PunctuatedAssigner 
    extends AssignerWithPunctuatedWatermarks[SensorReading] {

  val bound: Long = 60 * 1000 // 1 min in ms

  override def checkAndGetNextWatermark(
      r: SensorReading, 
      extractedTS: Long): Watermark = {
    if (r.id == "sensor_1") {
      // emit watermark if reading is from sensor_1
      new Watermark(extractedTS - bound)
    } else {
      // do not emit a watermark
      null
    }
  }

  override def extractTimestamp(
      r: SensorReading, 
      previousTS: Long): Long = {
    // assign record timestamp
    r.timestamp
  }
}

Tumbling Windows

A tumbling window assigner places elements to non-overlapping, fixed-size windows, as shown in the Figure Figure 6-1.

Figure 6-1. A Tumbling windows assigner places elements into fixed-size, non-overlapping windows.

The Datastream API provides two assigners, TumblingEventTimeWindows and TumblingProcessingTimeWindows for tumbling event-time and processing-time windows, respectively. A tumbling windows assigner receives one parameter which is the window size in time units and can be specified using the of(Time size) method of the assigner. The time interval can be set in milliseconds, seconds, minutes, hours, or days.

Example 6-12 and Example 6-13 show how to define event-time and processing-time tumbling windows on a stream of sensor data measurements.

val sensorData: DataStream[SensorReading] = ...

val avgTemp = sensorData
.keyBy(_.id)
// group readings in 1s event-time windows
.window(TumblingEventTimeWindows.of(Time.seconds(1)))
.process(new TemperatureAverager)

val avgTemp = sensorData
.keyBy(_.id)
// group readings in 1s processing-time windows
.window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
.process(new TemperatureAverager)

If you remember this example when we first encountered it in Chapter 2, the window definition looked a bit different. Back then, we defined an event-time tumbling window using the timeWindow(size) method, which is a shortcut for window.(TumblingEventTimeWindows.of(size)) or window.(TumblingProcessingTimeWindows.of(size)) depending on the configured time characteristic.

val avgTemp = sensorData
.keyBy(_.id)
// shortcut for window.(TumblingEventTimeWindows.of(size))
.timeWindow(Time.seconds(1))
.process(new TemperatureAverager)

By default, tumbling windows are aligned to the epoch time, i.e., 1970-01-01-00:00:00.000. For example, an assigner with a size of one hour will define windows at 00:00:00, 01:00:00, 02:00:00 and so on. Alternatively, you can specify an offset as a second parameter in the assigner. The example in Example 6-15 shows windows with an offset of 15 minutes that start at 00:15:00, 01:15:00, 02:15:00, etc.

val avgTemp = sensorData
.keyBy(_.id)
// group readings in 1 hour windows with 15 min offset
.window(TumblingEventTimeWindows.of(Time.hours(1), Time.minutes(15)))
.process(new TemperatureAverager

Sliding Windows

The sliding window assigner places stream elements to possibly overlapping, fixed-size windows, as shown in Figure Figure 6-2.

Figure 6-2. A Sliding windows assigner places elements into fixed-size, possibly overlapping windows.

For a sliding window, you have to specify a window size and a slide interval that defines how frequently a new window is started. When the slide interval is smaller than the window size, the windows overlap and elements be assigned to more than one window. If the slide is larger than the window size, some elements might not be assigned to any window and hence be dropped.

Example 6-16 shows how to group the sensor readings in sliding windows of 1 hour and 15 minutes slide. Each reading will be added to four windows. The DataStream API provides event-time and processing-time assigners, as well as shortcut methods, while a time interval offset can be set as the third parameter to the window assigner.

// event-time sliding windows assigner
val slidingAvgTemp = sensorData
.keyBy(_.id)
// create 1h event-time windows every 15 minutes
.window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(15)))
.process(new TemperatureAverager)


// processing-time sliding windows assigner
val slidingAvgTemp = sensorData
.keyBy(_.id)
// create 1h processing-time windows every 15 minutes
.window(SlidingProcessingTimeWindows.of(Time.hours(1), Time.minutes(15)))
.process(new TemperatureAverager)

// sliding windows assigner using a shortcut method
val slidingAvgTemp = sensorData
.keyBy(_.id)
// shortcut for window.(TumblingEventTimeWindows.of(size))
.timeWindow(Time.hours(1), Time(minutes(15)))
.process(new TemperatureAverager

Session Windows

A session window assigner places elements into non-overlapping windows of activity that have no fixed size. The boundaries of a session windows are defined by gaps of inactivity, i.e., time intervals in which no record is received. Figure Figure 6-3illustrates how elements are assigned to session windows.

Figure 6-3. A Session windows assigner places elements into varying-size, windows defined by a session gap.

The following examples show how to group the sensor readings into session windows where a session is defined by a 15 min period of inactivity:

// event-time session windows assigner
val sessionWindows = sensorData
.keyBy(_.id)
// create event-time session windows with a 15 min gap
.window(EventTimeSessionWindows.withGap(Time.minutes(15)))
.process(...)

// processing-time session windows assigner
val sessionWindows = sensorData
.keyBy(_.id)
// create processing-time session windows with a 15 min gap
.window(ProcessingTimeSessionWindows.withGap(Time.minutes(15)))
.process(...)

Since session windows do not have predefined start and end timestamps, a window assigner cannot immediately assign them to the correct window. Therefore, the SessionWindows assigner initially maps each incoming element into its own window with the element’s timestamp as the start time and the session gap as the window size. Subsequently, it merges all windows with overlapping ranges

ReduceFunction

The ReduceFunction was introduced in Chapter 5 when discussing running aggregations on keyed streams. A ReduceFunction accepts two values of the same type and combines them into a single value of the same type. When being applied on a windowed stream, a ReduceFunction incrementally aggregates the elements that are assigned to a window. A window only stores the current result of the aggregation, i.e., a single value of the ReduceFunction’s input (and output) type. When a new element is received, the ReduceFunction is called with the new element and the result that is read from the window’s state. The window’s state is replaced by the ReduceFunction’s result.

The advantages of applying a ReduceFunction on a window is the constant and small state size per window and the simple function interface. However, the applications for a ReduceFunction are limited and usually restricted to simple aggregations since the input and output type must be the same.

val minTempPerWindow: DataStream[(String, Double)] = sensorData
    .map(r => (r.id, r.temperature))
    .keyBy(_._1)
    .timeWindow(Time.seconds(15))
    .reduce((r1, r2) => (r1._1, r1._2.min(r2._2)))

For Example 6-18, we use a lambda function to specify how two elements of a window can be combined to produce an output of the same type. The same example can also be implemented with a class that implements the ReduceFunction interface as shows in Example 6-19.

val minTempPerWindow: DataStream[(String, Double)] = sensorData
  .map(r => (r.id, r.temperature))
  .keyBy(_._1)
  .timeWindow(Time.seconds(15))
  .reduce(new MinTempFunction)

// ================ //
// A reduce function to compute the minimum temperature per sensor.
class MinTempFunction extends ReduceFunction[(String, Double)] {
  override def reduce(r1: (String, Double), r2: (String, Double)) = {
    (r1._1, r1._2.min(r2._2))
  }
}

AggregateFunction

Similar to a ReduceFunction, an AggregateFunction is also incrementally applied to the elements that are applied to a window. Moreover, also the state of a window with an AggregateFunction consists of a single value.

However, the interface of the AggregateFunction is much more flexible but also more complex to implement compared to the interface of the ReduceFunction. Example 6-20 shows the interface of the AggregateFunction.

public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {

 // create a new accumulator to start a new aggregate.
 ACC createAccumulator();

 // add an input element to the accumulator and return the accumulator.
 ACC add(IN value, ACC accumulator);

 // compute the result from the accumulator and return it.
 OUT getResult(ACC accumulator);

 // merge two accumulators and return the result.
 ACC merge(ACC a, ACC b);
}

The interface defines a type for input elements, IN, an accumulator of type ACC, and a result type OUT. In contrast to the ReduceFunction, the intermediate data type and the output type do not depend on the input type.

val avgTempPerWindow: DataStream[(String, Double)] = sensorData
  .map(r => (r.id, r.temperature))
  .keyBy(_._1)
  .timeWindow(Time.seconds(15))
  .aggregate(new AvgTempFunction)

// ========= //

// An AggregateFunction to compute the average tempeature per sensor.
// The accumulator holds the sum of temperatures and an event count.
class AvgTempFunction
    extends AggregateFunction[(String, Double), (String, Double, Int), (String, Double)] {
  
  override def createAccumulator() = {
    ("", 0.0, 0)
  }

  override def add(in: (String, Double), acc: (String, Double, Int)) = {
    (in._1, in._2 + acc._2, 1 + acc._3)
  }

  override def getResult(acc: (String, Double, Int)) = {
    (acc._1, acc._2 / acc._3)
  }

  override def merge(acc1: (String, Double, Int), acc2: (String, Double, Int)) = {
    (acc1._1, acc1._2 + acc2._2, acc1._3 + acc2._3)
  }
}

ProcessWindowFunction

ReduceFunction and AggregateFunction are incrementally applied on events that are assigned to a window. However, sometimes we need access to all elements of a window to perform more complex computations, such as computing the the median of values in a window or the most frequently occurring value. For such applications, neither the ReduceFunction nor the AggregateFunction are suitable. Flink’s DataStream API offers the ProcessWindowFunction to perform arbitrary computations on the contents of a window.

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> 
    extends AbstractRichFunction {

  // Evaluates the window.
  void process(KEY key, Context ctx, Iterable<IN> vals, Collector<OUT> out) throws Exception;

  // Deletes any custom per-window state when the window is purged.
  public void clear(Context ctx) throws Exception {}

  // The context holding window metadata.
  public abstract class Context implements Serializable {
      
    // Returns the metadata of the window
    public abstract W window();

    // Returns the current processing time.
    public abstract long currentProcessingTime();

    // Returns the current event-time watermark.
    public abstract long currentWatermark();

    // State accessor for per-window state.
    public abstract KeyedStateStore windowState();

    // State accessor for per-key global state.
    public abstract KeyedStateStore globalState();

    // Emits a record to the side output identified by the OutputTag.
    public abstract <X> void output(OutputTag<X> outputTag, X value);
  }
}

The process() method is called with the key of the window, an Iterable to access the elements of the window, and a Collector to emit results. Moreover, the method has a Context parameter similar to other process methods. The Context object of the ProcessWindowFunction gives access to meta data of the window, the current processing time and watermark, state stores to manage per-window and per-key global state, as well as side outputs to emit records.

We already discussed some of the features of the Context object when introducing the ProcessFunction, such as access to the current processing and event time and side outputs. However, ProcessWindowFunction’s Context object also offers a unique feature. The meta data of the window typically contains information that can be used as an identifier for a window, such as the start and end timestamps in case of a time window.

Another feature are per-window and per-key global state. Global state refers to the keyed state that is not scoped to any window, while per-window state refers to the window instance that is currently being evaluated. Per-window state is useful to maintain information that should be shared between multiple invocations of the process() method on the same window, which can happen due to configuring allowed lateness or using a custom Trigger. A ProcessWindowFunction that utilizes per-window state needs to implement its clear() method to clean up any window-specific state before the window is purged. Global state can be used to share information between multiple windows on the same key.

// output the lowest and highest temperature reading every 5 seconds
val minMaxTempPerWindow: DataStream[MinMaxTemp] = sensorData
  .keyBy(_.id)
  .timeWindow(Time.seconds(5))
  .process(new HighAndLowTempProcessFunction)

// ========= //

case class MinMaxTemp(id: String, min: Double, max:Double, endTs: Long)

/**
 * A ProcessWindowFunction that computes the lowest and highest temperature
 * reading per window and emits a them together with the 
 * end timestamp of the window.
 */
class HighAndLowTempProcessFunction
    extends ProcessWindowFunction[SensorReading, MinMaxTemp, String, TimeWindow] {

  override def process(
      key: String,
      ctx: Context,
      vals: Iterable[SensorReading],
      out: Collector[MinMaxTemp]): Unit = {

    val temps = vals.map(_.temperature)
    val windowEnd = ctx.window.getEnd

    out.collect(MinMaxTemp(key, temps.min, temps.max, windowEnd))
  }
}

Internally, a window that is evaluated by ProcessWindowFunction stores all assigned events in a ListState¹. By collecting all events and providing access to window meta data and other features, the ProcessWindowFunction can address many more use cases than a ReduceFunction or AggregateFunction. However, the state of a window that collects all events can become significantly larger than the state of a window whose elements are incrementally aggregated. 

Incremental Aggregation and ProcessWindowFunction

The ProcessWindowFunction is a very powerful window function but you need to use it with caution since it typically holds more data in state than incrementally aggregating functions. In fact it is quite common, that most of the logic that needs be applied on a window can be expressed as an incremental aggregation but it also need access to window metadata or state.

In such a case, you can combine a ReduceFunction or AggregateFunction, which performs incremental aggreagtion, with a ProcessWindowFunction that provides access to more functionality. Elements that are assigned to a window will be immediately processed and when the Trigger of the window fires, the aggregated result will be handed to the ProcessWindowFunction. The Iterable parameter of the ProcessWindowFunction.process() method will only provide a single value, the incrementally aggregated result.

In the DataStream API this is done by providing a ProcessWindowFunction as a second parameter to the reduce() or aggregate() methods as shown in Example 6-24 and Example 6-25.

input
  .keyBy(...)
  .timeWindow(...)
  .reduce(incrAggregator: ReduceFunction[IN],
    function: ProcessWindowFunction[IN, OUT, K, W])

input
  .keyBy(...)
  .timeWindow(...)
  .aggregate(incrAggregator: AggregateFunction[IN, ACC, V],
    windowFunction: ProcessWindowFunction[V, OUT, K, W])

The example in Example 6-26 shows how to solve the same use case as the code in Example 6-23 with a combination of a ReduceFunction and a ProcessWindowFunction, i.e., how to emit every 5 seconds the minimun and maximum temperature per sensor and the end timestamp of each window.

val minMaxTempPerWindow2: DataStream[MinMaxTemp] = sensorData
  .map(r => (r.id, r.temperature, r.temperature))
  .keyBy(_._1)
  .timeWindow(Time.seconds(5))
  .reduce(
    // incrementally compute min and max temperature
    (r1: (String, Double, Double), r2: (String, Double, Double)) => {
      (r1._1, r1._2.min(r2._2), r1._3.max(r2._3))
    },
    // finalize result in ProcessWindowFunction
    new AssignWindowEndProcessFunction()
  )

// ========= //

case class MinMaxTemp(id: String, min: Double, max:Double, endTs: Long)

class AssignWindowEndProcessFunction
  extends ProcessWindowFunction[(String, Double, Double), MinMaxTemp, String, TimeWindow] {

  override def process(
      key: String,
      ctx: Context,
      minMaxIt: Iterable[(String, Double, Double)],
      out: Collector[MinMaxTemp]): Unit = {

    val minMax = minMaxIt.head
    val windowEnd = ctx.window.getEnd
    out.collect(MinMaxTemp(key, minMax._2, minMax._3, windowEnd))
  }
}

Window Lifecycle

A window operator creates and typically also deletes windows while it processes incoming stream elements. As discussed before, elements are assigned to windows by a WindowAssigner, a Trigger decides when to evalute a window, and a window function performs the actual window evaluation. In this section, we discuss the lifecycle of a window, i.e., when it is created, what information it consists of, and when it is deleted.

A window is created when a WindowAssigner assigns the first element to it. Consequently, there is no window without at least one element. A window consists of different pieces of state.

• The window content. The window content holds the elements that have been assigned to the window or the result of the incremental aggregation in case the window operator has a ReduceFunction or AggregateFunction.
• The window object. A WindowAssigner returns one, none, or multiple window objects. The window operator groups elements based on the returned objects. Hence, a window object holds the information to distinguish windows from each other. Each window object has an end timestamp that defines the point in time after which the window can be deleted.
• Timers of a Trigger. A Trigger can register timers to be called back at certain points in time, for example to evaluate a window or purge its content. These timers are maintained by the window operator.
• Custom-defined state in a Trigger. A Trigger can define and use custom, per-window and per-key state. This state is completely controlled by the trigger and not maintained by the window operator.

The window operator deletes a window, when the end time of the window, defined by the end timestamp of the window object, is reached. Whether this happens with processing time or event time semantics, depends on the value returned by the WindowAssigner.isEventTime() method.

When a window is deleted, the window operator automatically clears the window content and discards the window object. Custom-defined trigger state and registered trigger timers are not cleared because this state is opaque to the window operator. Hence, a trigger must clear all of its state in the Trigger.clear() method to prevent leaking state.

Window Assigners

A WindowAssigner determines for each arriving element to which windows it is assigned. An element can be added to one, none, but also multiple windows.

public abstract class WindowAssigner<T, W extends Window> 
    implements Serializable {

  // Returns a collection of windows to which the element is assigned.
  public abstract Collection<W> assignWindows(
    T element, 
    long timestamp, 
    WindowAssignerContext context);

  // Returns the default Trigger of the WindowAssigner.
  public abstract Trigger<T, W> getDefaultTrigger(
    StreamExecutionEnvironment env);

  // Returns the TypeSerializer for the windows of this WindowAssigner.
  public abstract TypeSerializer<W> getWindowSerializer(
    ExecutionConfig executionConfig);

  // Indicates whether this assigner creates event-time windows.
  public abstract boolean isEventTime();

  // A context that gives access to the current processing time.
  public abstract static class WindowAssignerContext {

    // Returns the current processing time.
    public abstract long getCurrentProcessingTime();
  }
}

A WindowAssigner is typed to the type of the incoming elements and the type of the windows to which the elements are assigned. It also needs to provide a default Trigger that is used if no explicit trigger is specified.

The code in Example 6-29 creates a custom assigner for 30 seconds tumbling event-time windows.

/** A custom window that groups events into 30 second tumbling windows. */
class ThirtySecondsWindows
    extends WindowAssigner[Object, TimeWindow] {

  val windowSize: Long = 30 * 1000L

  override def assignWindows(
      o: Object,
      ts: Long,
      ctx: WindowAssigner.WindowAssignerContext): java.util.List[TimeWindow] = {

    // rounding down by 30 seconds
    val startTime = ts - (ts % windowSize)
    val endTime = startTime + windowSize
    // emitting the corresponding time window
    Collections.singletonList(new TimeWindow(startTime, endTime))
  }

  override def getDefaultTrigger(
      env: environment.StreamExecutionEnvironment): Trigger[Object, TimeWindow] = {
    EventTimeTrigger.create()
  }

  override def getWindowSerializer(
      executionConfig: ExecutionConfig): TypeSerializer[TimeWindow] = {
    new TimeWindow.Serializer
  }

  override def isEventTime = true
}

In addition to the WindowAssigner interface there is also the MergingWindowAssigner interface that extends the WindowAssigner. The MergingWindowAssigner is used for window operators that need to merge existing windows. One example for such an assigner is the EventTimeSessionWindows assigner that we discussed before and which works by creating a new window for each arriving element and merging overlapping windows afterwards.

When merging windows, you need to ensure that the state of all merging windows and their triggers is also approriately merged. The Trigger interface features a callback method that is invoked when windows are merged to merge state that is associated to the windows. Merging of windows is discussed in more detail in the next section.

Triggers

Triggers define when a window is evaluated and its results are emitted. A trigger can decide to fire based on progress in time or data-specific conditions, such as element count or certain observed element values. For example, the default triggers of the previously discussed time-windows fire when the processing time or the watermark exceed the timestamp of the window’s end boundary.

Triggers have access to time properties, timers, and can work with state. Hence, they are similarly powerful as process functions. For example you can implement triggering logic to fire when the window received a certain number of elements, when an element with a specific value is added to the window, or after detecting a pattern on added elements like “two events of the same type within 5 seconds”. A custom trigger can also be used to compute and emit early results from an event-time window, i.e., before the watermark reaches the window’s end timestamp. This is a common strategy to produce (incomplete) low-latency results despite a conservative watermarking strategy.

Everytime a trigger is called it produces a TriggerResult that determines what should happend to the window. A TriggerResult can take one of the following values:

• CONTINUE: No action is taken.

• FIRE: If the window operator has a ProcessWindowFunction, the function is called and the result is emitted. If the window only has an incremetal aggregation function (ReduceFunction or AggregateFunction) the current aggregation result is emitted. The state of the window is not changed.

• PURGE: The content of the window is completely discarded and the window including all metadata is removed. Also the ProcessWindowFunction.clear() method is invoked to clean up all custom per-window state.

• FIRE_AND_PURGE: Evaluates the window first (FIRE) and subsequently removes all state and metadata (PURGE).

The possible TriggerResult values enable you to implement sophisticated windowing logic. A custom trigger may fire several times computing new or updated results or also purge a window without emitting a result if a certain condition is fulfilled.

Example 6-30shows the Trigger API.

public abstract class Trigger<T, W extends Window> implements Serializable {

  // Called for every element that gets added to a window.
  TriggerResult onElement(
    T element, long timestamp, W window, TriggerContext ctx);

  // Called when a processing-time timer fires.
  public abstract TriggerResult onProcessingTime(
    long timestamp, W window, TriggerContext ctx);

  // Called when an event-time timer fires.
  public abstract TriggerResult onEventTime(
    long timestamp, W window, TriggerContext ctx);

  // Returns true if this trigger supports merging of trigger state.
  public boolean canMerge();

  // Called when several windows have been merged into one window 
  // and the state of the triggers needs to be merged.
  public void onMerge(W window, OnMergeContext ctx);

  // Clears any state that the trigger might hold for the given window. 
  // This method is called when a window is purged.
  public abstract void clear(W window, TriggerContext ctx);
}

// A context object that is given to Trigger methods to allow them
// to register timer callbacks and deal with state.
public interface TriggerContext {

  // Returns the current processing time.
  long getCurrentProcessingTime();

  // Returns the current watermark time.
  long getCurrentWatermark();

  // Registers a processing-time timer.
  void registerProcessingTimeTimer(long time);

  // Registers an event-time timer
  void registerEventTimeTimer(long time);

  // Deletes a processing-time timer
  void deleteProcessingTimeTimer(long time);

  // Deletes an event-time timer.
  void deleteEventTimeTimer(long time);

  // Retrieves a state object that is scoped to the window and the key of the trigger.
  <S extends State> S getPartitionedState(StateDescriptor<S, ?> stateDescriptor);
}

// Extension of TriggerContext that is given to the Trigger.onMerge() method.
public interface OnMergeContext extends TriggerContext {
  .reduce/aggregate/process() //  define the window function
  // Merges per-window state of the trigger.
  // The state to be merged must support merging.
  void mergePartitionedState(StateDescriptor<S, ?> stateDescriptor);
}

As you see, the Trigger API can be used to implement sophisticated logic by providing access to time and state. There are two aspects about triggers that require special care, cleaning up state and merging triggers.

When using per-window state in a trigger, you need to ensure that this state is properly deleted when the window is deleted. Otherwise, the window operator will accumulate more and more state over time and your application will probably fail at some point in the future. In order to clean up all state when a window is deleted, the clear() method of a trigger needs to remove all custom per-window state and delete all processing-time and event-time timers using the TriggerContext object. It is not possible to clean up state in a timer callback method, since these methods are not called after a window was deleted.

If a trigger is applied together with a MergingWindowAssigner, it needs to be able to handle the situation when two windows are merged. In this case, also any custom state of their triggers need to be merged. The canMerge() declares that a trigger supports merging and the onMerge() method needs to implement the logic to perform the merge. If a trigger does not support merging it cannot be used in combination with a MergingWindowAssigner.

Merging of timers requires to provide the state descriptors of all custom state to the mergePartitionedState() method of the OnMergeContext object. Note that mergable triggers may only use state primitives that can be automatically merged, i.e., ListState, ReduceState, or AggregatingState.

/** A trigger that fires early. The trigger fires at most every second. */
class OneSecondIntervalTrigger
    extends Trigger[SensorReading, TimeWindow] {

  override def onElement(
      r: SensorReading,
      timestamp: Long,
      window: TimeWindow,
      ctx: Trigger.TriggerContext): TriggerResult = {

    // firstSeen will be false if not set yet
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean]("firstSeen", createTypeInformation[Boolean]))

    // register initial timer only for first element
    if (!firstSeen.value()) {
      // compute time for next early firing by rounding watermark to second
      val t = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
      ctx.registerEventTimeTimer(t)
      // register timer for the window end
      ctx.registerEventTimeTimer(window.getEnd)
      firstSeen.update(true)
    }
    // Continue. Do not evaluate per element
    TriggerResult.CONTINUE
  }

  override def onEventTime(
      timestamp: Long,
      window: TimeWindow,
      ctx: Trigger.TriggerContext): TriggerResult = {
    if (timestamp == window.getEnd) {
      // final evaluation and purge window state
      TriggerResult.FIRE_AND_PURGE
    } else {
      // register next early firing timer
      val t = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
      if (t < window.getEnd) {
        ctx.registerEventTimeTimer(t)
      }
      // fire trigger to evaluate window
      TriggerResult.FIRE
    }
  }

  override def onProcessingTime(
      timestamp: Long,
      window: TimeWindow,
      ctx: Trigger.TriggerContext): TriggerResult = {
    // Continue. We don't use processing time timers
    TriggerResult.CONTINUE
  }

  override def clear(
      window: TimeWindow,
      ctx: Trigger.TriggerContext): Unit = {

    // clear trigger state
    val firstSeen: ValueState[Boolean] = ctx.getPartitionedState(
      new ValueStateDescriptor[Boolean]("firstSeen", createTypeInformation[Boolean]))
    firstSeen.clear()
  }
}

Note that the trigger uses custom state which is cleaned up in the clear() method. Since we are using a simple non-mergable ValueState, the trigger is not mergable as well.

Evictors

The Evictor is an optional component in Flink’s windowing mechanism. It can remove elements from a window before or after the window function is evaluated. 

public interface Evictor<T, W extends Window> extends Serializable {

  // Optionally evicts elements. Called before windowing function.
  void evictBefore(
      Iterable<TimestampedValue<T>> elements, 
      int size, 
      W window, 
      EvictorContext evictorContext);

  // Optionally evicts elements. Called after windowing function.
  void evictAfter(
     Iterable<TimestampedValue<T>> elements, 
     int size, 
     W window, 
     EvictorContext evictorContext);

// A context object that is given to Evictor methods.
interface EvictorContext {

  // Returns the current processing time.
  long getCurrentProcessingTime();

  // Returns the current event time watermark.
  long getCurrentWatermark();
}

The evictBefore() and evictAfter() methods are called before and after a window function is applied on the content of a window, respectively. Both methods are called with an Iterable that serves all elements that were added to the window, the number of elements in the window (size), the window object, and an EvictorContext that provides access to the current processing time and watermark. Elements are removed from a window by calling the remove() method on the Iterator that can be obtained from the Iterable.

Evictors are often applied on a GlobalWindow for partial cleaning of the window, i.e., without purging the complete window state.