How to send statistics of a stream periodically to another stream in Flink

You can use the Side Outputs which

you can also produce any number of additional side output result streams. The type of data in the result streams does not have to match the type of data in the main stream and the types of the different side outputs can also differ. This operation can be useful when you want to split a stream of data where you would normally have to replicate the stream and then filter out from each stream the data that you don’t want to have.

Because the side output needs to extend a ProcessFunction or a KeyedProcessFunction you can leverage it to use the onTimer(). Here is one example.

ctx - An ProcessFunction.OnTimerContext that allows querying the timestamp of the firing timer, querying the TimeDomain of the firing timer and getting a TimerService for registering timers and querying the time. The context is only valid during the invocation of this method, do not store it.

The the gotcha is to use the out.collect(...) inside the processElement() method for every data that is consumed. And use the side output inside the onTimer() method to emit to a second stream every 5 minutes.

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