Apache Flink 测试中是否有虚拟时间的概念，就像在 Reactor 和 RxJava 中一样

Question

在 RxJava 和 Reactor 中，有虚拟时间的概念来测试依赖于时间的运算符。我不知道如何在 Flink 中执行此操作。例如，我将以下示例放在一起，我想在其中尝试处理迟到的事件以了解它们是如何处理的。但是我无法理解这样的测试会是什么样子？有没有办法结合 Flink 和 Reactor 让测试变得更好？

public class PlayWithFlink {

    public static void main(String[] args) throws Exception {

        final OutputTag<MyEvent> lateOutputTag = new OutputTag<MyEvent>("late-data"){};

        // TODO understand how BoundedOutOfOrderness is related to allowedLateness
        BoundedOutOfOrdernessTimestampExtractor<MyEvent> eventTimeFunction = new BoundedOutOfOrdernessTimestampExtractor<MyEvent>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(MyEvent element) {
                return element.getEventTime();
            }
        };

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<MyEvent> events = env.fromCollection(MyEvent.examples())
                .assignTimestampsAndWatermarks(eventTimeFunction);

        AggregateFunction<MyEvent, MyAggregate, MyAggregate> aggregateFn = new AggregateFunction<MyEvent, MyAggregate, MyAggregate>() {
            @Override
            public MyAggregate createAccumulator() {
                return new MyAggregate();
            }

            @Override
            public MyAggregate add(MyEvent myEvent, MyAggregate myAggregate) {
                if (myEvent.getTracingId().equals("trace1")) {
                    myAggregate.getTrace1().add(myEvent);
                    return myAggregate;
                }
                myAggregate.getTrace2().add(myEvent);
                return myAggregate;
            }

            @Override
            public MyAggregate getResult(MyAggregate myAggregate) {
                return myAggregate;
            }

            @Override
            public MyAggregate merge(MyAggregate myAggregate, MyAggregate acc1) {
                acc1.getTrace1().addAll(myAggregate.getTrace1());
                acc1.getTrace2().addAll(myAggregate.getTrace2());
                return acc1;
            }
        };

        KeySelector<MyEvent, String> keyFn = new KeySelector<MyEvent, String>() {
            @Override
            public String getKey(MyEvent myEvent) throws Exception {
                return myEvent.getTracingId();
            }
        };

        SingleOutputStreamOperator<MyAggregate> result = events
                .keyBy(keyFn)
                .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
                .allowedLateness(Time.seconds(20))
                .sideOutputLateData(lateOutputTag)
                .aggregate(aggregateFn);


        DataStream lateStream = result.getSideOutput(lateOutputTag);

        result.print("SessionData");

        lateStream.print("LateData");

        env.execute();
    }
}

class MyEvent {
    private final String tracingId;
    private final Integer count;
    private final long eventTime;

    public MyEvent(String tracingId, Integer count, long eventTime) {
        this.tracingId = tracingId;
        this.count = count;
        this.eventTime = eventTime;
    }

    public String getTracingId() {
        return tracingId;
    }

    public Integer getCount() {
        return count;
    }

    public long getEventTime() {
        return eventTime;
    }

    public static List<MyEvent> examples() {
        long now = System.currentTimeMillis();
        MyEvent e1 = new MyEvent("trace1", 1, now);
        MyEvent e2 = new MyEvent("trace2", 1, now);
        MyEvent e3 = new MyEvent("trace2", 1, now - 1000);
        MyEvent e4 = new MyEvent("trace1", 1, now - 200);
        MyEvent e5 = new MyEvent("trace1", 1, now - 50000);
        return Arrays.asList(e1,e2,e3,e4, e5);
    }

    @Override
    public String toString() {
        return "MyEvent{" +
                "tracingId='" + tracingId + '\'' +
                ", count=" + count +
                ", eventTime=" + eventTime +
                '}';
    }
}

class MyAggregate {
    private final List<MyEvent> trace1 = new ArrayList<>();
    private final List<MyEvent> trace2 = new ArrayList<>();


    public List<MyEvent> getTrace1() {
        return trace1;
    }

    public List<MyEvent> getTrace2() {
        return trace2;
    }

    @Override
    public String toString() {
        return "MyAggregate{" +
                "trace1=" + trace1 +
                ", trace2=" + trace2 +
                '}';
    }
}

运行的输出是：

SessionData:1> MyAggregate{trace1=[], trace2=[MyEvent{tracingId='trace2', count=1, eventTime=1551034666081}, MyEvent{tracingId='trace2', count=1, eventTime=1551034665081}]}
SessionData:3> MyAggregate{trace1=[MyEvent{tracingId='trace1', count=1, eventTime=1551034166081}], trace2=[]}
SessionData:3> MyAggregate{trace1=[MyEvent{tracingId='trace1', count=1, eventTime=1551034666081}, MyEvent{tracingId='trace1', count=1, eventTime=1551034665881}], trace2=[]}

但是我希望看到 e5 事件的 lateStream 触发器应该在第一个事件触发之前 50 秒。

Answer 1

如果您将水印分配器修改成这样

AssignerWithPunctuatedWatermarks eventTimeFunction = new AssignerWithPunctuatedWatermarks<MyEvent>() {
    long maxTs = 0;

    @Override
    public long extractTimestamp(MyEvent myEvent, long l) {
        long ts = myEvent.getEventTime();
        if (ts > maxTs) {
            maxTs = ts;
        }
        return ts;
    }

    @Override
    public Watermark checkAndGetNextWatermark(MyEvent event, long extractedTimestamp) {
        return new Watermark(maxTs - 10000);
    }
};

那么你会得到你期望的结果。我不推荐这个——只是用它来说明发生了什么。

这里发生的事情是 BoundedOutOfOrdernessTimestampExtractor 是一个周期性水印生成器，它只会每 200 毫秒（默认情况下）将水印插入到流中。因为您的作业在此之前很久就完成了，所以您的作业遇到的唯一水印是 Flink 在每个有限流的末尾注入的水印（值为 MAX_WATERMARK）。迟到是相对于水印而言的，你预计会迟到的事件却在水印之前到达。

通过切换到标点水印，您可以强制水印更频繁地出现，或者更精确地出现在流中的特定点。这通常是不必要的（而且过于频繁的水印会导致开销），但是当您想要对水印的顺序进行强有力的控制时会很有帮助。

至于怎么写测试，你可以看看 used in Flink's own tests, or at flink-spector。

更新：

与 BoundedOutOfOrdernessTimestampExtractor 关联的时间间隔是流的预期乱序程度的规范。在这个范围内到达的事件不被认为是迟到的，并且事件时间计时器在这个延迟结束之前不会触发，从而为乱序事件的到达留出时间。 allowedLateness 仅适用于 window API，并描述框架在正常 window 触发时间后保持 window 状态多长时间，以便事件仍可添加到 window 并导致延迟触发。在这个额外的时间间隔之后，window 状态被清除，后续事件被发送到侧输出（如果配置）。

所以当您使用 BoundedOutOfOrdernessTimestampExtractor<MyEvent>(Time.seconds(10)) 时，您 而不是 说 "wait 10 seconds after every event in case earlier events might still arrive"。但是你是说你的事件最多应该有 10 秒的乱序。因此，如果您正在处理实时事件流，这意味着您将等待最多 10 秒，以防更早的事件到达。（而且，如果您正在处理历史数据，那么您可能能够在 1 秒内处理 10 秒的数据，也可能不会——知道您将等待 n 秒的事件时间过去并不能说明实际需要多长时间。 )

有关此主题的更多信息，请参阅 Event Time and Watermarks。

Apache Flink 测试中是否有虚拟时间的概念，就像在 Reactor 和 RxJava 中一样

Is there a notion of virtual time in Apache Flink tests like there is in Reactor and RxJava

java

rx-java

project-reactor

apache-flink

flink-streaming