I assume you are referring to elements output from finishBundle. The problem is that the current window is an input to WindowFn.assignWindows. The new window can depend on the timestamp, the element itself, and the original window. I'm not sure how many users rely on this, however it has been part of our public windowing API for a long time, so I would guess that some users do use this in their WindowFns.
Reuven On Mon, May 4, 2020 at 11:48 AM Jan Lukavský <[email protected]> wrote: > There was a mention in some other thread, that in order to make user > experience as predictable as possible, we should try to make windows > idempotent, and once window is assigned, it should be never changed (and > timestamp move outside of the scope of window, unless a different windowfn > is applied). Because all Beam window functions are actually time based, and > output timestamp is known, what is the issue of applying windowfn to > elements output from @FinishBundle and assign the windows automatically? > On 5/4/20 8:07 PM, Reuven Lax wrote: > > This should not affect the ability of the user to specify the output > timestamp. Today FinishBundleContext.output forces you to specify the > window as well as the timestamp, which is a bit awkward. (I believe that it > also lets you create brand new windows in finishBundle, which is > interesting, but I'm not quite sure of the use case). > > On Mon, May 4, 2020 at 10:29 AM Robert Bradshaw <[email protected]> > wrote: > >> This is a really nice idea. Would the user still need to specify the >> timestamp of the output? I'm a bit ambivalent about calling it >> multiple times if OuptutReceiver alone is in the parameter list; this >> might not be obvious and could be surprising behavior. >> >> On Mon, May 4, 2020 at 10:13 AM Reuven Lax <[email protected]> wrote: >> > >> > I would like to discuss a minor extension to the Beam model. >> > >> > Beam bundles have very few restrictions on what can be in a bundle, in >> particular s bundle might contain records for many different windows. This >> was an explicit decision as bundling primarily exists for performance >> reasons and we found that limiting bundling based on windows or timestamps >> often led to severe performance problems. However it sometimes makes >> finishBundle hard to use. >> > >> > I've seen multiple cases where users maintain some state in their DoFn >> that needs finalizing (e.g. writing to an external service) in >> finishBundle. Often users end up keeping lists of all windows seen in the >> bundle so they can be processed separately (or sometimes not realizing that >> their can be multiple windows and writing incorrect code). >> > >> > The lack of a window also means that we don't currently support >> injecting an OuptutReceiver into finishBundle, as there's no good way of >> knowing which window output should be put into. >> > >> > I would like to propose adding a way for finishBundle to inspect the >> window, either directly (via a BoundedWindow parameter) or indirectly (via >> an OutputReceiver parameter). In this case, we will execute finishBundle >> once per window in the bundle. Otherwise, we will execute finishBundle once >> at the end of the bundle as before. This behavior is backwards compatible, >> as previously these parameters were disallowed in finishBundle. >> > >> > Note that this is similar to something Beam already does in >> processElement. A single element can exist in multiple windows, however if >> the processElement "observes" the window then Beam will call processElement >> once per window. >> > >> > In Java, the user code could look like this: >> > >> > DoFn<> { >> > ... >> > @FinishBundle >> > public void finishBundle(IntervalWindow window, OutputReceiver<T> o) >> { >> > // This finishBundle will be called once per window in the >> bundle since it has >> > // a parameter that observes the window. >> > } >> > } >> > >> > This PR shows an implementation of this extension for the Java SDK. >> > >> > Thoughts? >> > >> > Reuven >> >
