Flink基础（九）：DS简介(9) 基于时间和窗口的操作符（一）

在本章，我们将要学习DataStream API中处理时间和基于时间的操作符，例如窗口操作符。

首先，我们会学习如何定义时间属性，时间戳和水位线。然后我们将会学习底层操作process function，它可以让我们访问时间戳和水位线，以及注册定时器事件。接下来，我们将会使用Flink的window API，它提供了通常使用的各种窗口类型的内置实现。我们将会学到如何进行用户自定义窗口操作符，以及窗口的核心功能：assigners（分配器）、triggers（触发器）和evictors（清理器）。最后，我们将讨论如何基于时间来做流的联结查询，以及处理迟到事件的策略。

1 设置时间属性

如果我们想要在分布式流处理应用程序中定义有关时间的操作，彻底理解时间的语义是非常重要的。当我们指定了一个窗口去收集某1分钟内的数据时，这个长度为1分钟的桶中，到底应该包含哪些数据？在DataStream API中，我们将使用时间属性来告诉Flink：当我们创建窗口时，我们如何定义时间。时间属性是StreamExecutionEnvironment的一个属性，有以下值：

ProcessingTime

机器时间在分布式系统中又叫做“墙上时钟”。

当操作符执行时，此操作符看到的时间是操作符所在机器的机器时间。Processing-time window的触发取决于机器时间，窗口包含的元素也是那个机器时间段内到达的元素。通常情况下，窗口操作符使用processing time会导致不确定的结果，因为基于机器时间的窗口中收集的元素取决于元素到达的速度快慢。使用processing time会为程序提供极低的延迟，因为无需等待水位线的到达。

如果要追求极限的低延迟，请使用processing time。

EventTime

当操作符执行时，操作符看的当前时间是由流中元素所携带的信息决定的。流中的每一个元素都必须包含时间戳信息。而系统的逻辑时钟由水位线(Watermark)定义。我们之前学习过，时间戳要么在事件进入流处理程序之前已经存在，要么就需要在程序的数据源（source）处进行分配。当水位线宣布特定时间段的数据都已经到达，事件时间窗口将会被触发计算。即使数据到达的顺序是乱序的，事件时间窗口的计算结果也将是确定性的。窗口的计算结果并不取决于元素到达的快与慢。

当水位线超过事件时间窗口的结束时间时，窗口将会闭合，不再接收数据，并触发计算。

IngestionTime

当事件进入source操作符时，source操作符所在机器的机器时间，就是此事件的“摄入时间”（IngestionTime），并同时产生水位线。IngestionTime相当于EventTime和ProcessingTime的混合体。一个事件的IngestionTime其实就是它进入流处理器中的时间。

IngestionTime没什么价值，既有EventTime的执行效率（比较低），有没有EventTime计算结果的准确性。

下面的例子展示了如何设置事件时间。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment;
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<SensorReading> sensorData = env.addSource(...);

如果要使用processing time，将TimeCharacteristic.EventTime替换为TimeCharacteristic.ProcessingTIme就可以了。

1.1 指定时间戳和产生水位线

如果使用事件时间，那么流中的事件必须包含这个事件真正发生的时间。使用了事件时间的流必须携带水位线。

时间戳和水位线的单位是毫秒，记时从1970-01-01T00:00:00Z开始。到达某个操作符的水位线就会告知这个操作符：小于等于水位线中携带的时间戳的事件都已经到达这个操作符了。时间戳和水位线可以由SourceFunction产生，或者由用户自定义的时间戳分配器和水位线产生器来生成。

Flink暴露了TimestampAssigner接口供我们实现，使我们可以自定义如何从事件数据中抽取时间戳。一般来说，时间戳分配器需要在source操作符后马上进行调用。

因为时间戳分配器看到的元素的顺序应该和source操作符产生数据的顺序是一样的，否则就乱了。这就是为什么我们经常将source操作符的并行度设置为1的原因。

也就是说，任何分区操作都会将元素的顺序打乱，例如：并行度改变，keyBy()操作等等。

所以最佳实践是：在尽量接近数据源source操作符的地方分配时间戳和产生水位线，甚至最好在SourceFunction中分配时间戳和产生水位线。当然在分配时间戳和产生水位线之前可以对流进行map和filter操作是没问题的，也就是说必须是窄依赖。

以下这种写法是可以的。

DataStream<T> stream = env
  .addSource(...)
  .map(...)
  .filter(...)
  .assignTimestampsAndWatermarks(...)

下面的例子展示了首先filter流，然后再分配时间戳和水位线。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment;

// 从调用时刻开始给env创建的每一个stream追加时间特征
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<SensorReading> readings = env
  .addSource(new SensorSource)
  .filter(r -> r.temperature > 25)
  .assignTimestampsAndWatermarks(new MyAssigner());

MyAssigner有两种类型

• AssignerWithPeriodicWatermarks
• AssignerWithPunctuatedWatermarks

以上两个接口都继承自TimestampAssigner。

1.2 周期性的生成水位线

周期性的生成水位线：系统会周期性的将水位线插入到流中（水位线也是一种特殊的事件!）。默认周期是200毫秒，也就是说，系统会每隔200毫秒就往流中插入一次水位线。

这里的200毫秒是机器时间！

可以使用ExecutionConfig.setAutoWatermarkInterval()方法进行设置。

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// 每隔5秒产生一个水位线
env.getConfig.setAutoWatermarkInterval(5000)

上面的例子产生水位线的逻辑：每隔5秒钟，Flink会调用AssignerWithPeriodicWatermarks中的getCurrentWatermark()方法。如果方法返回的时间戳大于之前水位线的时间戳，新的水位线会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于等于之前水位线的时间戳，则不会产生新的水位线。

例子，自定义一个周期性的时间戳抽取

scala version

class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading] {
  val bound = 60 * 1000 // 延时为1分钟
  var maxTs = Long.MinValue + bound + 1 // 观察到的最大时间戳

  override def getCurrentWatermark: Watermark {
    new Watermark(maxTs - bound - 1)
  }

  override def extractTimestamp(r: SensorReading, previousTS: Long) {
    maxTs = maxTs.max(r.timestamp)
    r.timestamp
  }
}

java version

.assignTimestampsAndWatermarks(
  // generate periodic watermarks
  new AssignerWithPeriodicWatermarks[(String, Long)] {
    val bound = 10 * 1000L // 最大延迟时间
    var maxTs = Long.MinValue + bound + 1 // 当前观察到的最大时间戳

    // 用来生成水位线
    // 默认200ms调用一次
    override def getCurrentWatermark: Watermark = {
      println("generate watermark!!!" + (maxTs - bound - 1) + "ms")
      new Watermark(maxTs - bound - 1)
    }

    // 每来一条数据都会调用一次
    override def extractTimestamp(t: (String, Long), l: Long): Long = {
      println("extract timestamp!!!")
      maxTs = maxTs.max(t._2) // 更新观察到的最大事件时间
      t._2 // 抽取时间戳
    }
  }
)

如果我们事先得知数据流的时间戳是单调递增的，也就是说没有乱序。我们可以使用assignAscendingTimestamps，方法会直接使用数据的时间戳生成水位线。

scala version

val stream = ...
val withTimestampsAndWatermarks = stream.assignAscendingTimestamps(e => e.timestamp)

java version

.assignTimestampsAndWatermarks(
        WatermarkStrategy
                .<SensorReading>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<SensorReading>() {
                    @Override
                    public long extractTimestamp(SensorReading r, long l) {
                        return r.timestamp;
                    }
                })
)

如果我们能大致估算出数据流中的事件的最大延迟时间，可以使用如下代码：

最大延迟时间就是当前到达的事件的事件时间和之前所有到达的事件中最大时间戳的差。

scala version

.assignTimestampsAndWatermarks(
  // 水位线策略；默认200ms的机器时间插入一次水位线
  // 水位线 = 当前观察到的事件所携带的最大时间戳 - 最大延迟时间
  WatermarkStrategy
    // 最大延迟时间设置为5s
    .forBoundedOutOfOrderness[(String, Long)](Duration.ofSeconds(5))
    .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
      // 告诉系统第二个字段是时间戳，时间戳的单位是毫秒
      override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = element._2
    })
)

java version

.assignTimestampsAndWatermarks(
        WatermarkStrategy
                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                        return element.f1;
                    }
                })
)

以上代码设置了最大延迟时间为5秒。

1.3 如何产生不规则的水位线

有时候输入流中会包含一些用于指示系统进度的特殊元组或标记。Flink为此类情形以及可根据输入元素生成水位线的情形提供了AssignerWithPunctuatedWatermarks接口。该接口中的checkAndGetNextWatermark()方法会在针对每个事件的extractTimestamp()方法后立即调用。它可以决定是否生成一个新的水位线。如果该方法返回一个非空、且大于之前值的水位线，算子就会将这个新水位线发出。

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading] {
  val bound = 60 * 1000

  // 每来一条数据就调用一次
  // 紧跟`extractTimestamp`函数调用
  override def checkAndGetNextWatermark(r: SensorReading, extractedTS: Long) {
    if (r.id == "sensor_1") {
      // 抽取的时间戳 - 最大延迟时间
      new Watermark(extractedTS - bound)
    } else {
      null
    }
  }

  // 每来一条数据就调用一次
  override def extractTimestamp(r: SensorReading, previousTS: Long) {
    r.timestamp
  }
}

现在我们已经知道如何使用 TimestampAssigner 来产生水位线了。现在我们要讨论一下水位线会对我们的程序产生什么样的影响。

水位线用来平衡延迟和计算结果的正确性。水位线告诉我们，在触发计算（例如关闭窗口并触发窗口计算）之前，我们需要等待事件多长时间。基于事件时间的操作符根据水位线来衡量系统的逻辑时间的进度。

完美的水位线永远不会错：时间戳小于水位线的事件不会再出现。在特殊情况下(例如非乱序事件流)，最近一次事件的时间戳就可能是完美的水位线。启发式水位线则相反，它只估计时间，因此有可能出错，即迟到的事件(其时间戳小于水位线标记时间)晚于水位线出现。针对启发式水位线，Flink提供了处理迟到元素的机制。

设定水位线通常需要用到领域知识。举例来说，如果知道事件的迟到时间不会超过5秒，就可以将水位线标记时间设为收到的最大时间戳减去5秒。另一种做法是，采用一个Flink作业监控事件流，学习事件的迟到规律，并以此构建水位线生成模型。

如果最大延迟时间设置的很大，计算出的结果会更精确，但收到计算结果的速度会很慢，同时系统会缓存大量的数据，并对系统造成比较大的压力。如果最大延迟时间设置的很小，那么收到计算结果的速度会很快，但可能收到错误的计算结果。不过Flink处理迟到数据的机制可以解决这个问题。上述问题看起来很复杂，但是恰恰符合现实世界的规律：大部分真实的事件流都是乱序的，并且通常无法了解它们的乱序程度(因为理论上不能预见未来)。水位线是唯一让我们直面乱序事件流并保证正确性的机制; 否则只能选择忽视事实，假装错误的结果是正确的。

• 思考题一：实时程序，要求实时性非常高，并且结果并不一定要求非常准确，那么应该怎么办？
• 回答：直接使用处理时间。
• 思考题二：如果要进行时间旅行，也就是要还原以前的数据集当时的流的状态，应该怎么办？
• 回答：使用事件时间。使用Hive将数据集先按照时间戳升序排列，再将最大延迟时间设置为0。

2 处理函数

我们之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下，极为重要。例如MapFunction这样的map转换算子就无法访问时间戳或者当前事件的事件时间。

基于此，DataStream API提供了一系列的Low-Level转换算子。可以访问时间戳、水位线以及注册定时事件。还可以输出特定的一些事件，例如超时事件等。Process Function用来构建事件驱动的应用以及实现自定义的业务逻辑(使用之前的window函数和转换算子无法实现)。例如，Flink-SQL就是使用Process Function实现的。

Flink提供了8个Process Function：

• ProcessFunction
• KeyedProcessFunction
• CoProcessFunction
• ProcessJoinFunction
• BroadcastProcessFunction
• KeyedBroadcastProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

我们这里详细介绍一下KeyedProcessFunction。

KeyedProcessFunction用来操作KeyedStream。KeyedProcessFunction会处理流的每一个元素，输出为0个、1个或者多个元素。所有的Process Function都继承自RichFunction接口，所以都有open()、close()和getRuntimeContext()等方法。而KeyedProcessFunction[KEY, IN, OUT]还额外提供了两个方法:

• processElement(v: IN, ctx: Context, out: Collector[OUT]), 流中的每一个元素都会调用这个方法，调用结果将会放在Collector数据类型中输出。Context可以访问元素的时间戳，元素的key，以及TimerService时间服务。Context还可以将结果输出到别的流(side outputs)。
• onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[OUT])是一个回调函数。当之前注册的定时器触发时调用。参数timestamp为定时器所设定的触发的时间戳。Collector为输出结果的集合。OnTimerContext和processElement的Context参数一样，提供了上下文的一些信息，例如firing trigger的时间信息(事件时间或者处理时间)。

2.1 时间服务和定时器

Context和OnTimerContext所持有的TimerService对象拥有以下方法:

• currentProcessingTime(): Long 返回当前处理时间
• currentWatermark(): Long 返回当前水位线的时间戳
• registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前key的processing time的timer。当processing time到达定时时间时，触发timer。
• registerEventTimeTimer(timestamp: Long): Unit 会注册当前key的event time timer。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
• deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。
• deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。

当定时器timer触发时，执行回调函数onTimer()。processElement()方法和onTimer()方法是同步（不是异步）方法，这样可以避免并发访问和操作状态。

针对每一个key和timestamp，只能注册一个定期器。也就是说，每一个key可以注册多个定时器，但在每一个时间戳只能注册一个定时器。KeyedProcessFunction默认将所有定时器的时间戳放在一个优先队列中。在Flink做检查点操作时，定时器也会被保存到状态后端中。

举个例子说明KeyedProcessFunction如何操作KeyedStream。

下面的程序展示了如何监控温度传感器的温度值，如果温度值在一秒钟之内(processing time)连续上升，报警。

scala version

val warnings = readings
  .keyBy(r => r.id)
  .process(new TempIncreaseAlertFunction)

  class TempIncrease extends KeyedProcessFunction[String, SensorReading, String] {
    // 懒加载；
    // 状态变量会在检查点操作时进行持久化，例如hdfs
    // 只会初始化一次，单例模式
    // 在当机重启程序时，首先去持久化设备寻找名为`last-temp`的状态变量，如果存在，则直接读取。不存在，则初始化。
    // 用来保存最近一次温度
    // 默认值是0.0
    lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(
      new ValueStateDescriptor[Double]("last-temp", Types.of[Double])
    )

    // 默认值是0L
    lazy val timer: ValueState[Long] = getRuntimeContext.getState(
      new ValueStateDescriptor[Long]("timer", Types.of[Long])
    )

    override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
      // 使用`.value()`方法取出最近一次温度值，如果来的温度是第一条温度，则prevTemp为0.0
      val prevTemp = lastTemp.value()
      // 将到来的这条温度值存入状态变量中
      lastTemp.update(value.temperature)

      // 如果timer中有定时器的时间戳，则读取
      val ts = timer.value()

      if (prevTemp == 0.0 || value.temperature < prevTemp) {
        ctx.timerService().deleteProcessingTimeTimer(ts)
        timer.clear()
      } else if (value.temperature > prevTemp && ts == 0) {
        val oneSecondLater = ctx.timerService().currentProcessingTime() + 1000L
        ctx.timerService().registerProcessingTimeTimer(oneSecondLater)
        timer.update(oneSecondLater)
      }
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
      out.collect("传感器ID是 " + ctx.getCurrentKey + " 的传感器的温度连续1s上升了！")
      timer.clear()
    }
  }

java version

DataStream<String> warings = readings
    .keyBy(r -> r.id)
    .process(new TempIncreaseAlertFunction());

看一下TempIncreaseAlertFunction如何实现, 程序中使用了ValueState这样一个状态变量, 后面会详细讲解。

    public static class TempIncreaseAlertFunction extends KeyedProcessFunction<String, SensorReading, String> {

        private ValueState<Double> lastTemp;
        private ValueState<Long> currentTimer;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            lastTemp = getRuntimeContext().getState(
                    new ValueStateDescriptor<>("last-temp", Types.DOUBLE)
            );
            currentTimer = getRuntimeContext().getState(
                    new ValueStateDescriptor<>("current-timer", Types.LONG)
            );
        }

        @Override
        public void processElement(SensorReading r, Context ctx, Collector<String> out) throws Exception {
            // 取出上一次的温度
            Double prevTemp = 0.0;
            if (lastTemp.value() != null) {
                prevTemp = lastTemp.value();
            }
            // 将当前温度更新到上一次的温度这个变量中
            lastTemp.update(r.temperature);

            Long curTimerTimestamp = 0L;
            if (currentTimer.value() != null) {
                curTimerTimestamp = currentTimer.value();
            }
            if (prevTemp == 0.0 || r.temperature < prevTemp) {
                // 温度下降或者是第一个温度值，删除定时器
                ctx.timerService().deleteProcessingTimeTimer(curTimerTimestamp);
                // 清空状态变量
                currentTimer.clear();
            } else if (r.temperature > prevTemp && curTimerTimestamp == 0) {
                // 温度上升且我们并没有设置定时器
                long timerTs = ctx.timerService().currentProcessingTime() + 1000L;
                ctx.timerService().registerProcessingTimeTimer(timerTs);
                // 保存定时器时间戳
                currentTimer.update(timerTs);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect("传感器id为: "
                    + ctx.getCurrentKey()
                    + "的传感器温度值已经连续1s上升了。");
            currentTimer.clear();
        }
    }

2.2 将事件发送到侧输出

大部分的DataStream API的算子的输出是单一输出，也就是某种数据类型的流。除了split算子，可以将一条流分成多条流，这些流的数据类型也都相同。process function的side outputs功能可以产生多条流，并且这些流的数据类型可以不一样。一个side output可以定义为OutputTag[X]对象，X是输出流的数据类型。process function可以通过Context对象发射一个事件到一个或者多个side outputs。

例子

scala version

object SideOutputExample {

  val output = new OutputTag[String]("side-output")

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new SensorSource)

    val warnings = stream
      .process(new FreezingAlarm)

    warnings.print() // 打印主流
    warnings.getSideOutput(output).print() // 打印侧输出流

    env.execute()
  }

  class FreezingAlarm extends ProcessFunction[SensorReading, SensorReading] {
    override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      if (value.temperature < 32.0) {
        ctx.output(output, "传感器ID为：" + value.id + "的传感器温度小于32度！")
      }
      out.collect(value)
    }
  }
}

java version

public class SideOutputExample {

    private static OutputTag<String> output = new OutputTag<String>("side-output"){};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<SensorReading> stream = env.addSource(new SensorSource());

        SingleOutputStreamOperator<SensorReading> warnings = stream
                .process(new ProcessFunction<SensorReading, SensorReading>() {
                    @Override
                    public void processElement(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
                        if (value.temperature < 32) {
                            ctx.output(output, "温度小于32度！");
                        }
                        out.collect(value);
                    }
                });

        warnings.print();
        warnings.getSideOutput(output).print();

        env.execute();
    }
}

2.3 CoProcessFunction

对于两条输入流，DataStream API提供了CoProcessFunction这样的low-level操作。CoProcessFunction提供了操作每一个输入流的方法: processElement1()和processElement2()。类似于ProcessFunction，这两种方法都通过Context对象来调用。这个Context对象可以访问事件数据，定时器时间戳，TimerService，以及side outputs。CoProcessFunction也提供了onTimer()回调函数。下面的例子展示了如何使用CoProcessFunction来合并两条流。

scala version

object SensorSwitch {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream = env.addSource(new SensorSource).keyBy(r => r.id)

    val switches = env.fromElements(("sensor_2", 10 * 1000L)).keyBy(r => r._1)

    stream
      .connect(switches)
      .process(new SwitchProcess)
      .print()

    env.execute()
  }

  class SwitchProcess extends CoProcessFunction[SensorReading, (String, Long), SensorReading] {

    lazy val forwardSwitch = getRuntimeContext.getState(
      new ValueStateDescriptor[Boolean]("switch", Types.of[Boolean])
    )

    override def processElement1(value: SensorReading, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      if (forwardSwitch.value()) {
        out.collect(value)
      }
    }

    override def processElement2(value: (String, Long), ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]): Unit = {
      forwardSwitch.update(true)
      ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + value._2)
    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#OnTimerContext, out: Collector[SensorReading]): Unit = {
      forwardSwitch.clear()
    }
  }
}

java version

public class SensorSwitch {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        KeyedStream<SensorReading, String> stream = env
                .addSource(new SensorSource())
                .keyBy(r -> r.id);

        KeyedStream<Tuple2<String, Long>, String> switches = env
                .fromElements(Tuple2.of("sensor_2", 10 * 1000L))
                .keyBy(r -> r.f0);

        stream
                .connect(switches)
                .process(new SwitchProcess())
                .print();

        env.execute();
    }

    public static class SwitchProcess extends CoProcessFunction<SensorReading, Tuple2<String, Long>, SensorReading> {

        private ValueState<Boolean> forwardingEnabled;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            forwardingEnabled = getRuntimeContext().getState(
                    new ValueStateDescriptor<>("filterSwitch", Types.BOOLEAN)
            );
        }

        @Override
        public void processElement1(SensorReading value, Context ctx, Collector<SensorReading> out) throws Exception {
            if (forwardingEnabled.value() != null && forwardingEnabled.value()) {
                out.collect(value);
            }
        }

        @Override
        public void processElement2(Tuple2<String, Long> value, Context ctx, Collector<SensorReading> out) throws Exception {
            forwardingEnabled.update(true);
            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + value.f1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<SensorReading> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            forwardingEnabled.clear();
        }
    }
}