对于实时的流式处理系统来说，我们需要关注数据输入、计算和输出的及时性，所以处理延迟是一个比较重要的监控指标，特别是在数据量大或者软硬件条件不佳的环境下。Flink早在FLINK-3660就为用户提供了开箱即用的链路延迟监控功能，只需要配置好metrics.latency.interval,参数，再观察TaskManagerJobMetricGroup/operator_id/operator_subtask_index/latency,这个metric即可。本文简单walk一下源码，看看它是如何实现的，并且简要说明注意事项。

LatencyMarker的产生

与通过水印来标记事件时间的推进进度相似，Flink也用一种特殊的流元素（StreamElement）作为延迟的标记，称为LatencyMarker。
image.png
LatencyMarker的数据结构甚简单，只有3个field，即它被创建时携带的时间戳、算子ID和算子并发实例（sub-task）的ID。

  1. private final long markedTime;
  2. private final OperatorID operatorId;
  3. private final int subtaskIndex;

LatencyMarker和水印不同，不需要通过用户抽取产生，而是在Source端自动按照metrics.latency.interval参数指定的周期生成。StreamSource专门实现了一个内部类LatencyMarksEmitter用来发射LatencyMarker，而它又借用了负责协调处理时间的服务ProcessingTimeService，如下代码所示。

  1. LatencyMarksEmitter<OUT> latencyEmitter = null;
  2. if (latencyTrackingInterval > 0) {
  3.     latencyEmitter = new LatencyMarksEmitter<>(
  4.         getProcessingTimeService(),
  5.         collector,
  6.         latencyTrackingInterval,
  7.         this.getOperatorID(),
  8.         getRuntimeContext().getIndexOfThisSubtask());
  9. }
  11. private static class LatencyMarksEmitter<OUT> {
  12.     private final ScheduledFuture<?> latencyMarkTimer;
  14.     public LatencyMarksEmitter(
  15.             final ProcessingTimeService processingTimeService,
  16.             final Output<StreamRecord<OUT>> output,
  17.             long latencyTrackingInterval,
  18.             final OperatorID operatorId,
  19.             final int subtaskIndex) {
  20.         latencyMarkTimer = processingTimeService.scheduleAtFixedRate(
  21.             new ProcessingTimeCallback() {
  22.                 @Override
  23.                 public void onProcessingTime(long timestamp) throws Exception {
  24.                     try {
  25.                         // ProcessingTimeService callbacks are executed under the checkpointing lock
  26.                         output.emitLatencyMarker(new LatencyMarker(processingTimeService.getCurrentProcessingTime(), operatorId, subtaskIndex));
  27.                     } catch (Throwable t) {
  28.                         // we catch the Throwables here so that we don't trigger the processing
  29.                         // timer services async exception handler
  30.                         LOG.warn("Error while emitting latency marker.", t);
  31.                     }
  32.                 }
  33.             },
  34.             0L,
  35.             latencyTrackingInterval);
  36.     }
  38.     public void close() {
  39.         latencyMarkTimer.cancel(true);
  40.     }
  41. }

通过调用Output.emitLatencyMarker()方法，LatencyMarker就会随着数据流一起传递到下游了。

LatencyMarker的粒度

AbstractStreamOperator是所有Flink Streaming算子的基类，在它的初始化方法setup()中，会先创建用于延迟统计的LatencyStats实例。

  1. final String configuredGranularity = taskManagerConfig.getString(MetricOptions.LATENCY_SOURCE_GRANULARITY);
  2. LatencyStats.Granularity granularity;
  3. try {
  4.     granularity = LatencyStats.Granularity.valueOf(configuredGranularity.toUpperCase(Locale.ROOT));
  5. } catch (IllegalArgumentException iae) {
  6.     granularity = LatencyStats.Granularity.OPERATOR;
  7.     LOG.warn(
  8.         "Configured value {} option for {} is invalid. Defaulting to {}.",
  9.         configuredGranularity,
  10.         MetricOptions.LATENCY_SOURCE_GRANULARITY.key(),
  11.         granularity);
  12. }
  13. TaskManagerJobMetricGroup jobMetricGroup = this.metrics.parent().parent();
  14. this.latencyStats = new LatencyStats(jobMetricGroup.addGroup("latency"),
  15.     historySize,
  16.     container.getIndexInSubtaskGroup(),
  17.     getOperatorID(),
  18.     granularity);

在创建LatencyStats之前，先要根据metrics.latency.granularity配置项来确定延迟监控的粒度，分为以下3档：

  • single：每个算子单独统计延迟；
  • operator（默认值）：每个下游算子都统计自己与Source算子之间的延迟；
  • subtask：每个下游算子的sub-task都统计自己与Source算子的sub-task之间的延迟。

一般情况下采用默认的operator粒度即可，这样在Sink端观察到的latency metric就是我们最想要的全链路（端到端）延迟，以下也是以该粒度讲解。subtask粒度太细，会增大所有并行度的负担，不建议使用。

LatencyMarker的流转与计量

AbstractStreamOperator分别提供了用于单输入流算子OneInputStreamOperator、双输入流算子TwoInputStreamOperator的LatencyMarker处理方法。

  1. // ------- One input stream
  2. public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
  3.     reportOrForwardLatencyMarker(latencyMarker);
  4. }
  6. // ------- Two input stream
  7. public void processLatencyMarker1(LatencyMarker latencyMarker) throws Exception {
  8.     reportOrForwardLatencyMarker(latencyMarker);
  9. }
  11. public void processLatencyMarker2(LatencyMarker latencyMarker) throws Exception {
  12.     reportOrForwardLatencyMarker(latencyMarker);
  13. }
  15. protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
  16.     // all operators are tracking latencies
  17.     this.latencyStats.reportLatency(marker);
  18.     // everything except sinks forwards latency markers
  19.     this.output.emitLatencyMarker(marker);
  20. }

这些方法都会做两件事，一是计算延时并报告给LatencyStats，二是继续将LatencyMarker发射到下游。不妨来看看RecordWriterOutput.emitLatencyMarker()方法的具体实现。

  1. @Override
  2. public void emitLatencyMarker(LatencyMarker latencyMarker) {
  3.     serializationDelegate.setInstance(latencyMarker);
  4.     try {
  5.         recordWriter.randomEmit(serializationDelegate);
  6.     }
  7.     catch (Exception e) {
  8.         throw new RuntimeException(e.getMessage(), e);
  9.     }
  10. }
  12. /**
  13.  * This is used to send LatencyMarks to a random target channel.
  14.  */
  15. public void randomEmit(T record) throws IOException, InterruptedException {
  16.     emit(record, rng.nextInt(numberOfChannels));
  17. }

可见是从该算子所有的输出channel中随机选择一条来发射LatencyMarker，这样在度量算子级别延迟的基础上不会造成LatencyMarker泛滥，同时也不会受到并行度调整（重新分区）的影响。

注意StreamSink的reportOrForwardLatencyMarker()方法不会再发射LatencyMarker（因为已经处理完了），只会更新延迟。

  1. @Override
  2. protected void reportOrForwardLatencyMarker(LatencyMarker marker) {
  3.     // all operators are tracking latencies
  4.     this.latencyStats.reportLatency(marker);
  5.     // sinks don't forward latency markers
  6. }

LatencyStats中的延迟最终会转化为直方图表示，通过直方图就可以统计出延时的最大值、最小值、均值、分位值（quantile）等指标。以下是reportLatency()方法的源码。

  1. public void reportLatency(LatencyMarker marker) {
  2.     final String uniqueName = granularity.createUniqueHistogramName(marker, operatorId, subtaskIndex);
  3.     DescriptiveStatisticsHistogram latencyHistogram = this.latencyStats.get(uniqueName);
  4.     if (latencyHistogram == null) {
  5.         latencyHistogram = new DescriptiveStatisticsHistogram(this.historySize);
  6.         this.latencyStats.put(uniqueName, latencyHistogram);
  7.         granularity.createSourceMetricGroups(metricGroup, marker, operatorId, subtaskIndex)
  8.             .addGroup("operator_id", String.valueOf(operatorId))
  9.             .addGroup("operator_subtask_index", String.valueOf(subtaskIndex))
  10.             .histogram("latency", latencyHistogram);
  11.     }
  12.     long now = System.currentTimeMillis();
  13.     latencyHistogram.update(now - marker.getMarkedTime());
  14. }

可见，延迟是由当前时间戳减去LatencyMarker携带的时间戳得到的，所以在Sink端统计到的就是全链路延迟了。

注意事项

由以上分析可知，LatencyMarker是不会像Watermark一样参与到数据流的用户逻辑中的，而是直接被各算子转发并统计。这如何能得到真正的延时呢？如果由于网络不畅、数据流量太大等原因造成了反压（back pressure，之后再提），那么LatencyMarker的流转就会被阻碍，传递到下游的时间差就会增加，所以还是能够近似估算出整体的延时的。为了让它尽量精确，有两点特别需要注意：

  • ProcessingTimeService产生时间戳最终是靠System.currentTimeMillis()方法，所以必须保证Flink集群内所有节点的时区、时间是同步的，可以用ntp等工具来配置。
  • metrics.latency.interval的时间间隔宜大不宜小，在我们的实践中一般配置成30000（30秒）左右。一是因为延迟监控的频率可以不用太频繁，二是因为LatencyMarker的处理也要消耗时间，只有在LatencyMarker的耗时远小于正常StreamRecord的耗时时，metric反映出的数据才贴近实际情况，所以LatencyMarker的密度不能太大。