https://flink.apache.org/ logo
Join Slack
Powered by
I might phrase this poorly, but I'm interested in ...
# troubleshooting
n
I might phrase this poorly, but I'm interested in understanding a detail of event time. If you partition a datastream by keys with 
keyby
, do the watermarks of those partitioned streams evolve separately from one another?
Motivation for this question is that I have one stream of input data that has information from multiple devices that will then be processed independently. I'd like to make sure the watermarking advances uniquely for these independent devices.
z
Hey @Nathanael England! When you partition a data stream by keys with the keyBy() operator in Apache Flink, the watermarks of the partitioned streams do evolve separately from one another. This is because the keyBy() operator creates a new stream for each unique key value, and each of these new streams has its own watermark generator, which is responsible for generating and emitting watermarks for the events in that stream.
When the watermark generator receives an event, it compares the event's timestamp to the current watermark and updates the watermark if the event's timestamp is greater than the current watermark. This means that the watermark of each partitioned stream will advance based on the timestamps of the events in that stream, rather than being based on the timestamps of all events in the original stream. In your case, if you use the keyBy() operator to partition your data stream by the device id, each partitioned stream will have its own watermark generator, and the watermarks will evolve independently for each device. This way you can have a separate watermark for each device based on the event time of that device. You can also use a custom watermark generator to create your own watermarking strategy, for example, based on the device id, or you can use the 
BoundedOutOfOrdernessTimestampExtractor
which allows you to specify a maximum out of orderness for events, this way you can configure a different watermark for each device.
n
I see, thanks for confirming. So since I know I want events from different devices to be processed independently, it sounds like 
keyby
is the right thing to do and that will handle things fine if one device is idle for a long time while other devices are sending data in happily. I'm not sure I'm understanding the implications raised in https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/event-time/generating_watermarks/#dealing-with-idle-sources, though.
✅ 1
z
IDK which part is not comfortable for you, but this linked part what you sent just describing a problem that can occur when using the WatermarkGenerator feature, where some input partitions may not have any new events for a period of time, known as an idle input or an idle source. This can cause the watermark to be held back, as it is computed based on the minimum of all parallel watermarks. To address this issue, the documentation suggests using a WatermarkStrategy that can detect idleness and mark an input as idle. The specific code example there is using the "forBoundedOutOfOrderness" method with a 20 second delay and a 1 minute idleness duration. This strategy will detect when an input has not provided any new events for 1 minute and mark it as idle, allowing the watermark to be generated correctly.
n
Sorry, yes that was vague from me. I'm not sure I'm grasping why it matters if a watermark is held back if there's no data. Is that just affecting windows triggering and such?
z
Yes, that's correct. The watermark plays an important role in determining when to trigger window operations in Apache Flink. A watermark is a timestamp that is used to define the "current" time in a stream, up to which all events have been processed. It is used to determine when a window of events is complete and can be processed.
In the case where some of the input partitions do not carry events for a while, the watermark will be held back because it is computed as the minimum over all the different parallel watermarks. This means that the watermark will be delayed and will not reflect the correct "current" time in the stream
As a result, windows that are supposed to trigger based on the watermark will also be delayed and may not trigger when they are supposed to. This can cause inaccuracies in the results of window operations and lead to stale or incomplete data being processed. Therefore, it's important to detect idleness and mark an input as idle in order to prevent this problem and ensure that the watermark is generated correctly and windows are triggered correctly.
n
Does a watermark becoming idle change the watermark value or does it just allow triggering of windows and such that depend on watermark evolution?
z
AFAIK An idle watermark does not change the watermark value itself, but it does change how the watermark is used to trigger certain operations such as windowing or state eviction.
An idle watermark is a watermark that is not advancing because the input stream is not generating new events. This means that the WatermarkGenerator is not getting any new information to base the watermark on. As a result, the watermark value will be held at its last assigned value, without advancing.
d
Maybe one clarification: with a keyBy, the watermarks in each of the parallel tasks are handled independently. But you don’t get per-key watermarking — the watermarking in each parallel instance is based on all of the events for all of the keys being handled by that instance.
n
What would be the right way to get per-key watermarking? Or is that an uncommon pattern?
d
There are certain use cases where it would be desirable, but per-key watermarking is not possible. If you really want those semantics, your best option might be to implement all of the timing logic in a process function, and not use watermarks at all.
n
How is something like this typically done, then? Say for two mobile devices where device A is offline and device B is online and you want to check them for some conditions. Normal watermarking/lateness seems like it would progress the watermark due to events processed from device B. Then A comes back online, sends over its data, but no processing happens because it's been so long and its event timestamps are so much older than the watermark.
d
Yes, that’s exactly the use case where per-key watermarking would be desirable. There isn’t an ideal workaround. One approach is what I mentioned above, using a KeyedProcessFunction. You use keyed state to maintain some application-specific data, and process events as they arrive. You can, for example, store the max timestamp encountered so far for each key, which is what per-key watermarking would do, and use that to decide how to process incoming events, and when to emit results. An alternative is to handle the not-late events normally, using watermarks, and implement special handling for late events. E.g., if batches of late events are relatively rare, you could spin up another instance of the job to handle each such batch. If each backfill job is only processing events for a single key, it won’t have any problems with watermarking.
5 Views
Open in Slack
PreviousNext
Powered by