如何定期将流的统计信息发送到 Flink 中的另一个流

Question

我有一个流应用程序，它从 kafka 主题读取数据并从文件中读取数据，聚合它们并创建结果。

每 5 分钟，我想获取消耗的记录数以及从文件中读取的记录数并将其发送到另一个流。

我该怎么做？

Answer 1

您可以使用Side Outputs 其中

您还可以生成任意数量的附加输出结果流。结果流中的数据类型不必与主流中的数据类型匹配，并且不同侧输出的类型也可以不同。当您想要拆分通常必须复制流的数据流，然后从每个流中过滤掉您不希望拥有的数据时，此操作会很有用。

因为侧面输出需要扩展ProcessFunction 或KeyedProcessFunction，您可以利用它来使用onTimer()。 Here is one example.

ctx - 一个 ProcessFunction.OnTimerContext 允许查询触发计时器的时间戳，查询触发计时器的 TimeDomain 并获取用于注册计时器和查询时间的 TimerService。上下文仅在该方法调用期间有效，请勿存储。

问题是在processElement() 方法中使用out.collect(...) 来处理每个消耗的数据。并使用 onTimer() 方法内的侧输出每 5 分钟发送到第二个流。

public class SideOutputWithTimer extends KeyedProcessFunction<Tuple, Tuple2<String, String>, Tuple2<String, Long>> {

    final OutputTag<Tuple2<String, Long>> outputTag = new OutputTag<Tuple2<String, Long>>("side-output") {
    };

    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);

        // emit data without transforming it
        out.collect(Tuple2.of(value.f0, 1L));
    }

    @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
        // USE 5 MINITES INSTEAD OF 60000 milliseconds
        if (timestamp == result.lastModified + 60000) {
            // emit data to side output on timeout and after aggregating it
            ctx.output(outputTag, Tuple2.of(result.key, result.count));
        }
    }
}

class CountWithTimestamp {

    public String key;
    public long count;
    public long lastModified;
}