Flink 使用介绍相关文档目录

Flink 使用介绍相关文档目录

前言

概括来说，watermark用于基于event time的流计算系统数据流可能发生乱序的情况。对于event time数据流，接收到数据的时间和上游产生数据的时间是不相关的，因此可能会出现产生时间较早的数据由于网络抖动等原因到达Flink系统较晚的情形。Watermark用于应对数据乱序的情况。Watermark是数据流中的一种特殊数据，由Flink内部周期（可自定义）产生。下游接收到watermark的时候，会认为timestamp在watermark之前的数据已经到齐。针对这些数据的运算过程可以开始。Watermark之后的数据没有到齐，需要在缓存的同时，等待后续数据的到来。Watermark的生成策略可以实现数据乱序的兼容。例如将watermark发送的时间设置为当前接收到的数据的最大timestamp（记为t）减去5s。这样下游认为t-5s之前的数据已经到齐。t-5s之后的数据先缓存起来等待。从而实现容忍“乱序的程度”不超过5s的情形。

更为详细的Flink watermark讲解参见以下文章：

https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/concepts/time/

Flink 源码之时间处理

接下来讲解如何配置使用watermark生成策略(WatermarkStrategy)。

配置watermark自动发送周期

Flink默认的watermark自动发送周期为200ms。Flink支持全局方式和局部方式配置自动发送周期。

全局配置方式：修改flink-conf.yaml文件，增加或修改
pipeline.auto-watermark-interval(对应PipelineOptions.AUTO_WATERMARK_INTERVAL)配置项。

作业局部修改方式：调用ExecutionConfig.setAutoWatermarkInterval(...)方法。

env.getConfig().setAutoWatermarkInterval(100);

WatermarkStrategy使用

forMonotonousTimestamps

生成单调递增的watermark。数据流元素到来的时候不发送watermark，仅在到达自动发送周期的时候才发送。

这里递增的意思并非仅仅是watermark时间戳数值的严格递增，每次发送的watermark都是最近接收到的元素携带的timestamp。（从元素提取出携带的timestamp过程由TimestampAssigner负责，后面分析）。一种例外情况是如果遇到迟到的数据（watermark比前一个元素小），这个元素的watermark会被MonotonousTimestamps排除不做记录，可以保证向下游发送的watermark是递增的。

// 使用env创建数据源
source = ...

source.assignTimestampsAndWatermarks(WatermarkStrategy.<元素类型>forMonotonousTimestamps();

forBoundedOutOfOrderness

上面的单调递增方式无法解决元素乱序的问题。这里的BoundedOutOfOrderness是专门为数据存在乱序这种场景考虑的。使用时候需要指定一个参数，即最大可容忍的数据迟到时间。如果乱序数据迟到超过这个时间限制，该数据将被忽略。当然还可以配置为旁路输出，参见Flink 使用之数据分流。

BoundedOutOfOrderness实现并不复杂，基本和上面单调递增的方式一致。区别是在周期发送watermark的时候，发送的watermark需要减去最大可容忍的数据迟到时间。从而实现了数据计算的触发时刻向后拖延，在拖延的时间段内“等待”乱序数据到来。

使用方法如下：

source.assignTimestampsAndWatermarks(WatermarkStrategy.<元素类型>forBoundedOutOfOrderness(Duration.ofSeconds(30));

forGenerator

前面两种watermark generator能够满足绝大多数使用场景。如果仍不能满足要求，Flink提供了创建自定义watermark generator的方式。

这里以Integer类型的数据源为例，说明自定义generator的写法。

.assignTimestampsAndWatermarks(WatermarkStrategy.forGenerator(new WatermarkGeneratorSupplier<Integer>() {
    @Override
    public WatermarkGenerator<Integer> createWatermarkGenerator(Context context) {
        return new WatermarkGenerator<Integer>() {
            @Override
            public void onEvent(Integer integer, long eventTimestamp, WatermarkOutput watermarkOutput) {
                
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput watermarkOutput) {

            }
        };
    }
}));

在自定义WatermarkGenerator的时候按需实现两个方法：

• onEvent：接收到元素的时候触发。参数分别为输入的元素，元素提取出来的timestamp（event time）和控制输出watermark的对象（后面解释）。
• onPeriodicEmit：到达自动发送watermark周期的时候触发。参数只有一个，和前面的相同。

WatermarkOutput用来像下游发送watermark，或者控制数据输出。它有三个方法：

• emitWatermark: 发送watermark到下游。
• markIdle: 标记output为空闲状态。
• markActive: 标记output为活动状态。如果output在空闲状态发送了watermark，也会自动标记为活动状态。

Flink中一个计算步骤可能有多个上游（双数据流或更多），计算步骤会考虑到所有上游的watermark。设想如果一个流一直不产生watermark，需要等待这个流的数据呢还是这个流目前就没有数据可以忽略？Flink不好判断。为了解决这个问题引入了idle（空闲）机制。如果一个数据源标记了空闲状态，下游计算的时候不会考虑这个数据源的watermark。未能正确处理数据源的idle状态会导致Flink整个计算过程的阻塞。务必要注意这一点。

noWatermarks

不生成任何watermark。

source.assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks());

withTimestampAssigner

该方法用来配置如何从元素中抽取出watermark。例如到来的数据包含数据生成时的timestamp，格式为Tuple2<String, Long>类型，值为("Hello", 1690437024)。我们可以获取元素的第二个字段作为Flink内部的timestamp使用。

withTimestampAssigner使用方法如下所示：

source.assignTimestampsAndWatermarks(WatermarkStrategy.<元素类型>forMonotonousTimestamps().withTimestampAssigner((element, timestamp) -> {
    // element: 到来的元素
    // timestamp：上游为元素指定的timestamp，通常为数据源产生的timestamp
    // 需要编写自己的抽取逻辑
    // 返回抽取出的timestamp
}));

withIdleness

上面forGenerator章节提到了idle问题。大家可能会问：有没有一种常见的策略可以自动标记idle状态？比如数据流持续一段时间没有数据到来的时候自动标记为idle状态。withIdleness正好是这种策略。它对watermark generator做了包装。用户在其中不需要再去编写何时标记idle的逻辑。

withIdleness的用法如下所示：

.assignTimestampsAndWatermarks(WatermarkStrategy.<元素类型>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withIdleness(Duration.ofSeconds(5)));

WatermarkStrategy源代码分析

forBoundedOutOfOrderness

static <T> WatermarkStrategy<T> forBoundedOutOfOrderness(Duration maxOutOfOrderness) {
    return (ctx) -> new BoundedOutOfOrdernessWatermarks<>(maxOutOfOrderness);
}

上面创建了BoundedOutOfOrdernessWatermarks。继续查看它的代码。分析内容在注释中。

@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** The maximum timestamp encountered so far. */
    // 暂存最大的timestamp
    // 发送的timestamp一定是递增（或者大小不变）的
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    // 最大可容忍的数据迟到的时间范围（乱序程度）
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

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

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        // 接收到元素的时候，更新maxTimestamp
        // 如果接收到迟到的元素（eventTimestamp比maxTimestamp小），忽略不更新，确保timestamp递增
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 在发送watermark时间周期到来的时候，发送watermark
        // 发送的watermark需要减去outOfOrdernessMillis
        // 含义是让下游认为maxTimestamp - outOfOrdernessMillis - 1之前的数据已经到齐
        // 只有认为到齐的数据参会参与计算，未到齐的数据会缓存等待
        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
    }
}

forMonotonousTimestamps

static <T> WatermarkStrategy<T> forMonotonousTimestamps() {
    return (ctx) -> new AscendingTimestampsWatermarks<>();
}

这里创建了AscendingTimestampsWatermarks。它继承了BoundedOutOfOrdernessWatermarks，代码如下所示。

@Public
public class AscendingTimestampsWatermarks<T> extends BoundedOutOfOrdernessWatermarks<T> {

    /** Creates a new watermark generator with for ascending timestamps. */
    public AscendingTimestampsWatermarks() {
        super(Duration.ofMillis(0));
    }
}

看到构造方法很容易明白，AscendingTimestampsWatermarks是一种不容忍任何数据迟到的BoundedOutOfOrdernessWatermarks。

withIdleness

default WatermarkStrategy<T> withIdleness(Duration idleTimeout) {
    checkNotNull(idleTimeout, "idleTimeout");
    checkArgument(
            !(idleTimeout.isZero() || idleTimeout.isNegative()),
            "idleTimeout must be greater than zero");
    return new WatermarkStrategyWithIdleness<>(this, idleTimeout);
}

继续查看WatermarkStrategyWithIdleness的createWatermarkGenerator方法：

@Override
public WatermarkGenerator<T> createWatermarkGenerator(
        WatermarkGeneratorSupplier.Context context) {
    return new WatermarksWithIdleness<>(
            baseStrategy.createWatermarkGenerator(context), idlenessTimeout);
}

创建出的watermark generator为WatermarksWithIdleness。该类使用了装饰器模式。在不改变原有watermark generator的基础之上增加了标记idle的能力。它有三个成员变量。

// 包装的watermark生成器
private final WatermarkGenerator<T> watermarks;
// idle定时器，用来判断是否idle
private final IdlenessTimer idlenessTimer;
// 状态标记，目前是否处于idle状态
private boolean isIdleNow = false;

继续分析onEvent和onPeriodicEmit方法：

@Override
public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
    // 调用被包装watermark generator的onEvent方法
    watermarks.onEvent(event, eventTimestamp, output);
    // 告知idlenessTimer有活动发生
    idlenessTimer.activity();
    // 标记空闲状态为false
    isIdleNow = false;
}

@Override
public void onPeriodicEmit(WatermarkOutput output) {
    if (idlenessTimer.checkIfIdle()) {
        // 检查空闲状态，如果为空闲
        if (!isIdleNow) {
            // 如果当前状态不是空闲
            // 说明刚从活动状态变为空闲状态
            // 标记为idle状态
            output.markIdle();
            // 记录空闲状态为true
            isIdleNow = true;
        }
    } else {
        // 如果不是idle状态，调用包装的watermark generator的onPeriodicEmit方法
        watermarks.onPeriodicEmit(output);
    }
}

最后的问题就是IdlenessTimer是怎么判断是否idle的。我们继续分析它的构造函数，activity方法和checkIfIdle方法。

IdlenessTimer(Clock clock, Duration idleTimeout) {
    // 获取时钟
    this.clock = clock;

    long idleNanos;
    // 将idle超时时间转换为纳秒保存
    try {
        idleNanos = idleTimeout.toNanos();
    } catch (ArithmeticException ignored) {
        // long integer overflow
        idleNanos = Long.MAX_VALUE;
    }

    this.maxIdleTimeNanos = idleNanos;
}

public void activity() {
    // 内部有个计数器，只要有活动，该计数器自增1
    // counter是long类型，即便是自增溢出了，也不会影响
    counter++;
}

public boolean checkIfIdle() {
    if (counter != lastCounter) {
        // lastCounter为最近一次counter计数
        // 如果不等，说明期间有活动
        // 这里不写if (counter > lastCounter)的原因是兼容counter溢出的情况
        // activity since the last check. we reset the timer
        // 更新lastCounter，重设计时器
        lastCounter = counter;
        startOfInactivityNanos = 0L;
        return false;
    } else // timer started but has not yet reached idle timeout
    if (startOfInactivityNanos == 0L) {
        // first time that we see no activity since the last periodic probe
        // begin the timer
        // 首次发现counter没有更新，即没有活动，启用计时器
        startOfInactivityNanos = clock.relativeTimeNanos();
        return false;
    } else {
        // 如果当前时间和计时器时间差超过maxIdleTimeNanos，说明处于空闲状态
        return clock.relativeTimeNanos() - startOfInactivityNanos > maxIdleTimeNanos;
    }
}
    

本博客为作者原创，欢迎大家参与讨论和批评指正。如需转载请注明出处。