Using Broadcast State To Force Window Closure Using Fake Messages

Question

Description:

Currently I am working on using Flink with an IOT setup. Essentially, devices are sending data such as (device_id, device_type, event_timestamp, etc) and I don't have any control over when the messages get sent. I then key the steam by device_id and device_type to preform aggregations. I would like to use event-time given that is ensures the timers which are set trigger in a deterministic nature given a failure. However, given that this isn't always a high throughput stream a window could be opened for a 10 minute aggregation period, but not have its next point come until approximately 40 minutes later. Although the calculation would aggregation would eventually be completed it would output my desired result extremely late.

So my work around for this is to create an additional external source that does nothing other than pump fake messages. By having these fake messages being pumped out in alignment with my 10 minute aggregation period, even if a device hadn't sent any data, the event time windows would have something to force the windows closed. The critical part here is to make it possible that all parallel instances / operators have access to this fake message because I need to close all the windows with this single fake message. I was thinking that Broadcast state might be the most appropriate way to accomplish this goal given: "Broadcast state is replicated across all parallel instances of a function, and might typically be used where you have two streams, a regular data stream alongside a control stream that serves rules, patterns, or other configuration messages." Quote Source

Questions:

Is broadcast state the best method for ensuring all parallel instances (e.g. windows) receive my fake messages?
Once the operators have access to this fake message via the broadcast state can this fake message then be used to advance the event time watermark?

Are you saying that there might be no events for any of the devices for ~40 minutes? I ask, because any event can advance the watermark and close a window -- it doesn't have to be an event for a specific device. — David Anderson
Hey @DavidAnderson, yes that is exactly what I am saying. It's possible to have the devices firing all day and then at the end of the day all the devices stop sending events. In hindsight I realize that this is a better example to explain. So now that the day is over and the window isn't closed, I won't see the aggregation until the next day when the devices start sending events again. So I am trying to determine if there is a way to force this window closed. Basically I want a way to say, hey no events have come in and so it's time for me to send a watermark. — C. Johnson

David Anderson David Anderson · Accepted Answer · 2019-12-13T19:00:50

You can make this work with broadcast state, along the lines you propose, but I'm not convinced it's the best solution.

In an ideal world I'd suggest you arrange for the devices to send occasional keepalive messages, but assuming that's not possible, I think a custom Trigger would work well here. You can extend the EventTimeTrigger so that in addition to the event time timer it creates via

ctx.registerEventTimeTimer(window.maxTimestamp());

you also create a processing time timer, as a fallback, and you FIRE the window if the window still exists when that processing time timer fires.

I'm recommending this approach because it's simpler and more directly addresses the specific need. With the broadcast state approach you'll have to introduce a source for these messages, add a broadcast state descriptor and stream, add special fake watermarks for the non-broadcast stream (set to Watermark.MAX_WATERMARK), connect the broadcast and non-broadcast streams and implement a BroadcastProcessFunction (that probably doesn't really do anything), etc. It's a lot of moving parts spread across several different operators.

Using Broadcast State To Force Window Closure Using Fake Messages

1 Answers