Hi Timo, Thanks for your answer. I think I wasn't clear enough in my initial message, so let me give more details.
The stream is not keyed by timestamp, it's keyed by a custom field (e.g., user-id) and then fed into a KeyedProcessFunction. I want to process all events for a given user in order, before sending them downstream for further processing in other operators. I don't want to hold events longer than needed, hence using the watermark to signal which events can be processed. I don't think your suggestion of using a ListState would work, because we would effectively have one list per user. That would imply (among other things) that an event could only be processed when a new event for the same user arrives, which would not only imply a (potentially) huge latency, but also data leakage. Not to mention that the events being sent could easily be considered late-events to the downstream operators. The idea of keying by timestamp was an "evolution" of the ListState suggestion, where events-to-be-later-processed would be kept sorted in the map (which is what would be keyed by timestamp). We could iterate the map to process the events, instead of fetching the full list and sorting it to process the events in order. I don't think this solves any of the problems mentioned above, so I think that mentioning it only raised confusion. Regarding global event-time order, that's not really what I'm after. I only need event-time order per key, but I want to process the event as soon as possible, constrained by knowing that it is "safe" to do so because no event with a smaller timestamp for this key is yet to come. So, rephrasing my question as I'm not sure that part was clear in the initial message, here is the idea: - keeping one priority queue (ordered by timestamp) in each KeyedProcessFunction instance. Therefore, each priority queue would store events for multiple keys. - when an event arrives, we push it to the queue and then process events (updating state and sending them downstream) while their timestamp is lower than the current watermark. The question is: - is this fault tolerant? The priority queue is not state that is managed by flink, but it should be recoverable on replay. - is it possible that the events I'm sending downstream become late-events for a different operator, for some reason? Will they always be sent before the watermark of the event that originated the processElement() call? - I would effectively be processing multiple elements (from multiple keys) in the same call to processElement(). Is there a way to access the state of different keys? This doesn't feel like the right approach. Is there an operator more suitable than a KeyedProcessFunction which would allow me to handle the state for multiple keys in this task manager? Should I register a timer to trigger on the event timestamp instead? I believe timers trigger on watermarks, so that could theoretically work, although it feels a little weird. After all, what I want is just to buffer events so that they are only processed when the watermark has caught up to them. Thanks Timo Walther <twal...@apache.org> escreveu no dia sexta, 30/04/2021 à(s) 12:05: > Hi Miguel, > > your initial idea sounds not too bad but why do you want to key by > timestamp? Usually, you can simply key your stream by a custom key and > store the events in a ListState until a watermark comes in. > > But if you really want to have some kind of global event-time order, you > have two choices: > > - either a single operator with parallelism 1 that performs the ordering > - or you send the every event to every operator using the broadcast > state pattern [1] > > It is guaranteed that watermark will reach the downstream operator or > sink after all events. Watermarks are synchronized across all parallel > operator instances. You can store a Map uncheckpointed by this means > that you have to ensure to initialize the map again during recovery. > > Regards, > Timo > > > [1] > > https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/state/broadcast_state.html > > On 30.04.21 11:37, Miguel Araújo wrote: > > Hi everyone, > > > > I have a KeyedProcessFunction whose events I would like to process in > > event-time order. > > My initial idea was to use a Map keyed by timestamp and, when a new > > event arrives, iterate over the Map to process events older than the > > current watermark. > > > > The issue is that I obviously can't use a MapState, as my stream is > > keyed, so the map would be scoped to the current key. > > Is using a "regular" (i.e., not checkpointed) Map an option, given that > > its content will be recreated by the replay of the events on a restart? > > Is it guaranteed that the watermark that triggered the processing of > > multiple events (and their subsequent push downstream) is not sent > > downstream before these events themselves? > > > > Thanks, > > Miguel > >
