Details
-
New Feature
-
Status: Open
-
P3
-
Resolution: Unresolved
-
None
-
None
-
None
Description
Custom WindowFns are user specified windowing strategies that allow users to write a function to dictate how windows are assigned and subsequently merged together (if appropriate).
Unlike "known" windows like the Global Window, Interval Windows, and Session Windows, these are not executed by the runner, but executed at the direction of the runner as a bundle.
In particular, Custom WindowFns are likely to be implemented as a special class of DoFn like DoFns and CombineFns and are to be appropriate for use in Windowing strategies for Windowing into, but with an AssignWindows and a MergeWindows function. They'd need to be passed to `beam.WindoInto` via the window package, with a new wrapper function, `window.New` to convert the WindowFn into something `beam.WindowInto` can accept. Likely that most windows will be "interval window" encoded like Fixed and Sliding windows unless otherwise specified. As assignment and merging behavior is handled by the WindowFn, they could
Assign Windows accepts a `beam.EventTime`, and an element, and produces a non-empty `[]beam.Window` for that element.
MergeWindows accepts a slice of `[]beam.Window`.
`beam.Window` is an interface and can be type asserted to a custom type. However, the coder for the given concrete implementation of beam.Window needs to be known, so it's likely that a slice of the concrete type in question is best accepted and returned instead. Alternatively, the implementer can weigh the merits of having a single method that serve to return the reflect.Type of the concrete window type to serve the inference.
The execution node for WindowInto in the exec package would need to be updated to execute WindowFn methods when directed by the runner.
How WindowInto gets executed presently in the Go Direct Runner. Likely needs to be extended to handle the custom assigns, and merging.
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/runtime/exec/window.go
The user frontend package would require support for specifying the custom windowfns, or a new frontend function ("CustomWindowInto" ?) would need to be defined. This is also where new user documentation on how to define Custom WindowFns would need to go.
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/windowing.go
Defines known windows and payloads. One specifically for custom window fns, handled like structural DoFns is likely needed here.
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/graph/window/fn.go
Defines known window coders (Global and Interval, you'd want to add a Custom one)
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/graph/coder/windows.go
Defines existing concrete implementations of beam.Window
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/graph/window/windows.go
For a testing example CustomWindowFn in Java see: https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/WindowTest.java#L718
The new nodes needs to be added to the graphx translation, and the inverse on the exec side.
https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/runtime/graphx/translate.go#L422 https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/runtime/exec/translate.go#L179
Everything must be documented and tested, minimally to replicate existing Fixed, Sliding, and Session windowing behaviour.
Attachments
Issue Links
- mentioned in
-
Page Loading...