How to join a realtime stream and a delayed stream by event time

user_4475676 0
Viewed 643

I am trying to join 2 streams using CoProcessFunction. One of the input stream A is generated in realtime. However the other input stream B is loaded by a daily scheduled job with 1 day delay, meaning the event that is put in the stream today always has event time from yesterday.

With that being said, stream B's watermark will always be ~1 day behind A's watermark, so I would imagine a lot of events from A will be buffered in the memory. I am wondering if there is a way to solve issue. Some additional background, both stream A and B are kinesis streams(I am using FlinkKinesisConsumer) with retention period = 7 days.

Thanks in advance!

flink-streaming apache-flink
3 Answers

If You worry about the memory when buffering multiple elements, then You should take look at various different state backends especially RocksDb. This way, state will be kept on disk and not in memory.

This should easily allow solving the problem You are facing with buffering elements for one day since the only thing that limits the state size is disk space available, which is generally cheap and shouldn't be a problem in most of the cases.

user_88990440

Since Flink 1.8.1 the Flink Kinesis consumer has supported event-time alignment (selectively reading from splits to make sure that the various consumers advance evenly in event time). See Event Time Alignment for Shard Consumers for details.

The Flink community is working on providing more general support for event time synchronization across sources so that it will become possible to efficiently implement event time joins in cases like this where there is significant event time misalignment between the sources. Right now the only solution (unless you are using Kinesis) is use Flink state to buffer the stream that is ahead, which can lead to very large checkpoints and significant backpressure.

The underpinnings for general-purpose event time alignment are being implemented as part of FLIP-27 / FLINK-10740, after which the sources will have to be re-worked to take advantage of this new mechanism.

user_20008230

I think the situation you are facing may not actually be a problem, assuming that your 'fast' stream comes in via something like Kafka which can act as a buffer and retain the messages. (If this is not the case, it should be straightforward to write to Kafka before joining to create this situation yourself)

Though I did not test it, I would expect the fast stream to be ingested until the buffers that are available inside Flink are full. At that point it will simply stop or slow the ingestion until the second slow stream comes in to 'clean up' all the messages that are waiting to be joined, after which the fast stream could start moving again.

Note that this may require the messages in both streams to come in roughly in the same order.

user_9837220
Related