Hello, I've been able to produce a (relatively) minimal pipeline which reproduces the behavior I've previously noticed in one of our production pipelines.
The minimal pipeline repo can be found here: https://github.com/azk/late-samples-state The actual pipeline is implemented in one big class here: https://github.com/azk/late-samples-state/blob/master/src/main/java/late/samples/state/LateSamplesState.java Basically, this pipeline consumes events from a pubsub topic, which are very simple json objects. The events are: * deserialized * each element's timestamp is extracted from the json payload * the elements are keyed and then windowed in a one-minute fixed window with an allowed lateness of 7 days, with processing time triggers and accumulating panes. The pipeline then branches into 2 branches: 1. A simple GroupByKey, followed by a logging step which records which panes were fired. 2. A stateful ParDo, which collects elements in a BagState, logs what is currently in the Bag and sets a one minute timer, which also logs the contents of the BagState. When running this pipeline in Dataflow, and publishing a few "very late" events through the script which is part of the repo: https://github.com/azk/late-samples-state/blob/master/scripts/publish_events.py , the following behavior is observed: 1. The very late elements never appear after the GroupBy step, and are dropped as expected. 2. The very late elements are recorded in the stateful transform and added to the BagState. 3. When the timer expires, the BagState is empty and the elements that were previously in the Bag seem to disappear. Is this an expected behavior or is there some subtle issue going on? I'd be more than happy to deep dive into this with anyone interested in looking into this behavior. Thank you, Amit. On Wed, Mar 6, 2019 at 11:14 PM Amit Ziv-Kenet <amit.zivke...@gmail.com> wrote: > Hi Kenn, > Thank you for the the explanation! > > These points make total sense, that's why I was surprised with the > observed behavior, which breaks at least points 3 and 4. > I'll try to extract and share a minimal working example which demonstrates > this behavior. > > Thank you, > Amit. > > On Wed, Mar 6, 2019 at 8:11 PM Kenneth Knowles <k...@apache.org> wrote: > >> What you describe is not expected. Here are the relevant points, I think: >> >> - A window is expired when the watermark is past the end of the window + >> allowed lateness >> - An element is droppable when it is associated to an expired window >> - All droppable elements should be dropped before reaching the stateful >> ParDo step >> - The state and processing time timers for a particular key+window pair >> are garbage collected when the window is expired, because it is known that >> nothing can cause that state to be read >> - A key+window state is not cleared until all event time timers have >> been dispatched >> >> Do these make sense? I'd love to see more detail of your pipeline code. >> One thing to note is that an element being behind the watermark doesn't >> really matter. What matters is how the watermark relates to its window. >> >> Kenn >> >> On Wed, Mar 6, 2019 at 4:28 AM Amit Ziv-Kenet <amit.zivke...@gmail.com> >> wrote: >> >>> Hello, >>> >>> I'm encountering an unexpected behavior in one of our pipeline, I hope >>> you might help me make sense of it. >>> >>> This is a streaming pipeline implemented with the Java SDK (2.10) and >>> running on Dataflow. >>> >>> * In the pipeline a FixedWindow is applied to the data with an allowed >>> lateness of a week (doesn't really matter). >>> * The windowed PCollection is then used in 2 separate branches: one >>> regular GroupByKey and and a second one which is Transform which makes have >>> use of State and Timers. >>> * In the stateful transform, incoming elements are added to a BagState >>> container until some condition on the elements is reached, and the elements >>> in the BagState are dispatched downstream. A timer makes sure that the >>> BagState is dispatched even if the condition is not met after some timeout >>> has expired. >>> >>> Occasionally very late data enters the pipeline, with timestamps older >>> than the allowed lateness. >>> In these cases, the GroupByKey transform behaves as expected, and there >>> aren't any panes with the late data output downstream. >>> In the stateful transform, on the other hand, I see that these late >>> elements are processed and added to the BagState, but at a later point in >>> time (e.g. when the timer is triggered) the elements "disappear" from the >>> BagState and are no longer observable. This can happen multiple times, i.e. >>> late elements added to the BagState and then disappearing at a later time. >>> >>> Is this an expected behavior? Is there some sort of late data "garbage >>> collection" which eventually removes late elements? >>> >>> Thank you for your help, I hope my description is clear enough. >>> >>> Regards, >>> Amit. >>> >>