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

Question

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

每5分钟，我想得到多少记录被消耗和记录从文件中读取的计数，并将其发送到另一个流。

我该怎么做？

Answer 1

您可以使用侧面输出

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

因为侧输出需要扩展ProcessFunction或KeyedProcessFunction，所以您可以利用它来使用onTimer（）。这是一个例子。

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

问题的关键是在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;
}