过程函数（低级算子操作）

ProcessFunction
低级联接
例
KeyedProcessFunction
计时器
- 容错
- 定时器合并

译者：flink.sojb.cn

ProcessFunction

这ProcessFunction是一个低级流处理算子操作，可以访问所有（非循环）流应用程序的基本构建块：

事件（流数据元）
state（容错，一致，仅在被Key化的数据流上）
定时器（事件时间和处理时间，仅限被Key化的数据流）

该ProcessFunction可被认为是一个FlatMapFunction可以访问Keys状态和定时器。它通过为输入流中接收的每个事件调用来处理事件。

对于容错状态，ProcessFunction可以访问Flink的被Keys化状态，可以通过其访问 RuntimeContext，类似于其他有状态函数可以访问被Keys化状态的方式。

定时器允许应用程序对处理时间和事件时间的变化作出反应。每次调用该函数processElement(...)都会获得一个Context对象，该对象可以访问数据元的事件时间戳和TimerService。的TimerService可用于注册为将来事件- /处理-时刻回调。达到计时器的特定时间时，将onTimer(...)调用该方法。在该调用期间，所有状态再次限定为创建计时器的键，允许计时器操纵被Keys化状态。

注意如果要访问被Keys化状态和计时器，则必须应用ProcessFunction被Key化的数据流：

stream.keyBy(...).process(new MyProcessFunction())

低级联接

要在两个输入上实现低级算子操作，应用程序可以使用CoProcessFunction。此函数绑定到两个不同的输入，并从两个不同的输入获取单个调用processElement1(...)和 processElement2(...)记录。

实现低级别连接通常遵循以下模式：

为一个输入（或两者）创建状态对象
从输入接收数据元时更新状态
从其他输入接收数据元后，探测状态并生成连接结果

例如，您可能会将客户数据关联金融交易，同时保持客户数据的状态。如果您在面对乱序事件时需要完全和确定性的连接，那么当客户数据流的水印已经超过该交易时，您可以使用计时器来评估和发出交易的连接。

例

以下示例维护每个键的计数，并在每分钟通过（事件时间）时发出键/计数对，而不更新该键：

count，key和last-modification-timestamp存储在a中ValueState，它由key隐式定义。
对于每个记录，ProcessFunction递增计数器并设置最后修改时间戳
该函数还会在未来一分钟内调度回调（在事件时间内）
在每次回调时，它会根据存储计数的最后修改时间检查回调的事件时间时间戳，如果匹配则发出键/计数（即，在该分钟内没有进一步更新）

注意这个简单的例子可以用会话窗口实现。我们ProcessFunction在这里用它来说明它提供的基本模式。

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction.Context;
import org.apache.flink.streaming.api.functions.ProcessFunction.OnTimerContext;
import org.apache.flink.util.Collector;
// the source data stream
DataStream<Tuple2<String, String>> stream = ...;
// apply the process function onto a keyed stream
DataStream<Tuple2<String, Long>> result = stream
    .keyBy(0)
    .process(new CountWithTimeoutFunction());
/**
 * The data type stored in the state
 */
public class CountWithTimestamp {
    public String key;
    public long count;
    public long lastModified;
}
/**
 * The implementation of the ProcessFunction that maintains the count and timeouts
 */
public class CountWithTimeoutFunction extends ProcessFunction<Tuple2<String, String>, Tuple2<String, Long>> {
    /** The state that is maintained by this process function */
    private ValueState<CountWithTimestamp> state;
    @Override
    public void open(Configuration parameters) throws Exception {
        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
    }
    @Override
    public void processElement(Tuple2<String, String> value, Context ctx, Collector<Tuple2<String, Long>> out)
            throws Exception {
        // retrieve the current count
        CountWithTimestamp current = state.value();
        if (current == null) {
            current = new CountWithTimestamp();
            current.key = value.f0;
        }
        // update the state's count
        current.count++;
        // set the state's timestamp to the record's assigned event time timestamp
        current.lastModified = ctx.timestamp();
        // write the state back
        state.update(current);
        // schedule the next timer 60 seconds from the current event time
        ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out)
            throws Exception {
        // get the state for the key that scheduled the timer
        CountWithTimestamp result = state.value();
        // check if this is an outdated timer or the latest timer
        if (timestamp == result.lastModified + 60000) {
            // emit the state on timeout
            out.collect(new Tuple2<String, Long>(result.key, result.count));
        }
    }
}

import org.apache.flink.api.common.state.ValueState
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.functions.ProcessFunction.Context
import org.apache.flink.streaming.api.functions.ProcessFunction.OnTimerContext
import org.apache.flink.util.Collector
// the source data stream val stream: DataStream[Tuple2[String, String]] = ...
// apply the process function onto a keyed stream val result: DataStream[Tuple2[String, Long]] = stream
  .keyBy(0)
  .process(new CountWithTimeoutFunction())
/**
  * The data type stored in the state
  */
case class CountWithTimestamp(key: String, count: Long, lastModified: Long)
/**
  * The implementation of the ProcessFunction that maintains the count and timeouts
  */
class CountWithTimeoutFunction extends ProcessFunction[(String, String), (String, Long)] {
  /** The state that is maintained by this process function */
  lazy val state: ValueState[CountWithTimestamp] = getRuntimeContext
    .getState(new ValueStateDescriptor[CountWithTimestamp](a71b3b76b196cc32ecc6bcd3538f490e))
  override def processElement(value: (String, String), ctx: Context, out: Collector[(String, Long)]): Unit = {
    // initialize or retrieve/update the state 
    val current: CountWithTimestamp = state.value match {
      case null =>
        CountWithTimestamp(value._1, 1, ctx.timestamp)
      case CountWithTimestamp(key, count, lastModified) =>
        CountWithTimestamp(key, count + 1, ctx.timestamp)
    }
    // write the state back
    state.update(current)
    // schedule the next timer 60 seconds from the current event time
    ctx.timerService.registerEventTimeTimer(current.lastModified + 60000)
  }
  override def onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[(String, Long)]): Unit = {
    state.value match {
      case CountWithTimestamp(key, count, lastModified) if (timestamp == lastModified + 60000) =>
        out.collect((key, count))
      case _ =>
    }
  }
}

注意：在Flink 1.4.0之前，当从处理时间计时器调用时，该ProcessFunction.onTimer()方法将当前处理时间设置为事件时间时间戳。此行为非常微妙，用户可能不会注意到。嗯，这是有害的，因为处理时间时间戳是不确定的，不与水印对齐。此外，用户实现的逻辑依赖于这个错误的时间戳，很可能是出乎意料的错误。所以我们决定解决它。升级到1.4.0后，使用此不正确的事件时间戳的Flink作业将失败，用户应将其作业调整为正确的逻辑。

KeyedProcessFunction

KeyedProcessFunction作为其扩展ProcessFunction，可以在其onTimer(...) 方法中访问计时器的键。

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

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

计时器

两种类型的计时器（处理时间和事件时间）都由内部维护TimerService并排队执行。

在TimerService每个键和时间戳，即删除重复数据计时器，还有每键和时间戳最多一个计时器。如果为同一时间戳注册了多个计时器，则只onTimer()调用一次该方法。

注意 Flink同步onTimer()和的调用processElement()。因此，用户不必担心并发修改状态。

容错

定时器具有容错能力，并且与应用程序的状态一起检查点。如果故障恢复或从保存点启动应用程序，则会恢复计时器。

注意在恢复之前应该点火的检查点处理时间计时器将立即触发。当应用程序从故障中恢复或从保存点启动时，可能会发生这种情况。

注意除了RocksDB后台/增量SNAPSHOT/基于堆的定时器（将与之解决FLINK-10026）的组合之外，定时器始终是异步检查点。请注意，大量的计时器可以增加检查点时间，因为计时器是检查点状态的一部分。有关如何Reduce定时器数量的建议，请参阅“定时器合并”部分。

定时器合并

由于Flink每个键和时间戳只保存一个计时器，因此可以通过降低计时器分辨率来合并它们来Reduce计时器的数量。

对于1秒的定时器分辨率（事件或处理时间），您可以将目标时间向下舍入为完整秒数。定时器最多提前1秒发射，但不迟于要求，精确到毫秒。结果，每个键最多有一个计时器，第二个。

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

val coalescedTime = ((ctx.timestamp + timeout) / 1000) * 1000
ctx.timerService.registerProcessingTimeTimer(coalescedTime)

由于事件时间计时器仅在出现水印的情况下触发，因此您还可以使用当前的水印来计划和合并这些计时器和下一个水印：

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

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

也可以按如下方式停止和删除计时器：

停止处理时间计时器：

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

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

停止事件时间计时器：

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

val timestampOfTimerToStop = ...
ctx.timerService.deleteEventTimeTimer(timestampOfTimerToStop)

注意如果没有注册具有给定时间戳的此类计时器，则停止计时器无效。

过程函数（低级 算子操作）

ProcessFunction

低级联接

例

KeyedProcessFunction

计时器

容错

定时器合并