[
https://issues.apache.org/jira/browse/BEAM-11100?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Robert Burke updated BEAM-11100:
--------------------------------
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.
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.
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.
was:
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.
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.
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.
> [Go SDK] Support custom windowFns
> ---------------------------------
>
> Key: BEAM-11100
> URL: https://issues.apache.org/jira/browse/BEAM-11100
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
> Reporter: Robert Burke
> Priority: P3
>
> 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.
> 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.
> 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.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
