Flink中流动的四种元素

在Flink DataStream中流动着不同的元素，统称为StreamElement，StreamElement可以是StreamRecord、Watermark、StreamStatus、LatencyMarker中任何一种类型， S
treamElement
是一个抽象类，其他四种类型继承 S
treamElement
。

StreamElement

public abstract class StreamElement {

 //判断是否是Watermark

  public final boolean isWatermark() {

    return getClass() == Watermark.class;

  }

//判断是否为StreamStatus

  public final boolean isStreamStatus() {

    return getClass() == StreamStatus.class;

  }

//判断是否为StreamRecord

  public final boolean isRecord() {

    return getClass() == StreamRecord.class;

  }




//判断是否为LatencyMarker

  public final boolean isLatencyMarker() {

    return getClass() == LatencyMarker.class;

  }

  //转换为StreamRecord

  public final  StreamRecord asRecord() {

    return (StreamRecord) this;

  }




 //转换为Watermark

  public final Watermark asWatermark() {

    return (Watermark) this;

  }

//转换为StreamStatus

  public final StreamStatus asStreamStatus() {

    return (StreamStatus) this;

  }




//转换为LatencyMarker

  public final LatencyMarker asLatencyMarker() {

    return (LatencyMarker) this;

  }

}

StreamRecord

StreamRecord包含value、timestamp、hasTimestamp三个属性，其中value 可以
理解为我们处理的业务数据，
timestamp
表示数据的时间，也就是eventTime， hasTimestamp
标识是否具有timestamp。

public final class StreamRecord<T> extends StreamElement {




  /** The actual value held by this record. */

  private T value;




  /** The timestamp of the record. */

  private long timestamp;




  /** Flag whether the timestamp is actually set. */

  private boolean hasTimestamp;

  ...}

Watermark

Watermark在eventTime语义中产生的一类数据类型，内部包含一个时间戳timestamp，标识当前数据的时间进度。

public final class Watermark extends StreamElement {




  /** The watermark that signifies end-of-event-time. */

  public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);




  // ------------------------------------------------------------------------




  /** The timestamp of the watermark in milliseconds. */

  private final long timestamp;




  /**

   * Creates a new watermark with the given timestamp in milliseconds.

   */

  public Watermark(long timestamp) {

    this.timestamp = timestamp;

  }

.../

StreamStatus

先了解一下StreamStatus结构：

public final class StreamStatus extends StreamElement {




  public static final int IDLE_STATUS = -1; //闲置状态

  public static final int ACTIVE_STATUS = 0; //激活状态




  public static final StreamStatus IDLE = new StreamStatus(IDLE_STATUS);

  public static final StreamStatus ACTIVE = new StreamStatus(ACTIVE_STATUS);




  public final int status;

假设一个任务开启了两个并行度读取kafka数据的任务，每一个并行度都被分配了一个topic的partition，下游是一个事件时间窗口操作，那么窗口的触发与上游两个输入都有关系，如果现在出现了topic的一个partition没有数据了，那么就会导致窗口的一个输入没有数据，那么其watermark也就不会改变，根据watermark对齐机制可知，在window端处理的watermark取最小值，也就是其watermark不会改变，导致窗口也就永远无法触发，StreamStatus就是为了解决这种机制而产生的。

StreamS
tatus
有ACTIVE、IDEL两种状态， AC
TIVE
表示激活状态， IDEL
表示闲置状态，如果一个没有数据产生或者是在一个指定的时间内没有数据产生那么就会发送 IDEL
状态，通常是在Source端发送的，例如kafka线程没有分配到partition就会发送
，如下

//FlinkKafkaConsumerBase run方法中

if (subscribedPartitionsToStartOffsets.isEmpty()) {

      sourceContext.markAsTemporarilyIdle();

    }

那么在发送了IDLE_STATUS后，下游在做watermark对齐就不会考虑该输入的watermark，能够正常处理watermark完成对齐机制进而触发相应的窗口计算。

如果过了一段时间之后该stream又产生数据了，表示被激活了就会发送ACTIVE_STATUS，在下游参与对齐机制，如果这个stream的时间戳与另外一个stream的时间戳相差很大，会以该 strea
m
产生的数据时间作为参与watermark取值机制吗？当然不会，会拿 该
strea
m
产生的时间与当前的watermakr进行比较，必须大于等于当前的watermark才能参与对齐机制。

//StatusWatermarkValve inputStreamStatus方法中

if (channelStatuses[channelIndex].watermark >= lastOutputWatermark) {

        channelStatuses[channelIndex].isWatermarkAligned = true;

      }

LatencyMarker

LatencyMarker表示的是延时标记，其携带了一个时间戳，由source端周期性发出，用于统计由source端到下游operator所需耗时，通常用于延时测试使用，该标记在默认情况下是关闭的，可通过配置metrics.latency.interval开启，表示产生 LatencyMarker
的周期。

public final class LatencyMarker extends StreamElement {

  /** The time the latency mark is denoting. */

  private final long markedTime;




  private final OperatorID operatorId;




  private final int subtaskIndex;

  .....