apache / apache/beam

[Go SDK] Support custom windowFns

Open
#20,627 0 comments 0 reactions 0 assignees View on GitHub
go new feature P3
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
2d 2h
Merged PRs (30d)
205

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](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](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](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](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](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](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/graphx/translate.go#L422)  [https://github.com/apache/beam/blob/6e7b1c44bc7275ee047afc059fd610cd3f4e5bee/sdks/go/pkg/beam/core/runtime/exec/translate.go#L179](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.

Imported from Jira [BEAM-11100](https://issues.apache.org/jira/browse/BEAM-11100). Original Jira may contain additional context.
Reported by: lostluck.

Contributor guide

Open the contributing guide

Research direction

Start by reading sdks/go/pkg/beam/core/runtime/exec/window.go and sdks/go/pkg/beam/pkg/beam/windowing.go, then compare the Java CustomWindowFn example in WindowTest.java. Trace the graphx and exec translation points, window function and coder definitions, and existing window implementations. Done means custom assignment and merging work end to end, with documentation and tests covering Fixed, Sliding, and Session-like behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
data-engineering, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.