apache / apache/beam

[Feature Request][Go SDK]: Enable emitting to windowed PCollection from FinishBundle()

Open
#23,963 2 comments 0 reactions 0 assignees View on GitHub
bug go new feature P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### What would you like to happen?

I originally asked this question on [StackOverflow](https://stackoverflow.com/questions/74028678/using-finishbundle-in-apache-beam-go-sdk), but it does appear that there's a gap between python/java functionality and Go. According to this [documentation](https://beam.apache.org/documentation/transforms/python/elementwise/pardo/), if emitting data into a PCollection from `finish_bundle()`, you must use a `WindowedValue`.

This concept doesn't exist for the Go SDK and any data that's emitted from `FinishBundle()` is [emitted into the `SingleGlobalWindow`](https://github.com/apache/beam/blob/85b6b643bff1157fa68787ab9bd6b0afcd6c8bc6/sdks/go/pkg/beam/core/runtime/exec/pardo.go#L234). I see a [TODO comment](https://github.com/apache/beam/blob/85b6b643bff1157fa68787ab9bd6b0afcd6c8bc6/sdks/go/pkg/beam/core/runtime/exec/pardo.go#L125) in the `StartBundle()` method that refers to a resolved [Jira Issue BEAM-3303](https://issues.apache.org/jira/browse/BEAM-3303), but I don't see any existing Github Issues that address this gap.

The below isn't a functioning example, but is essentially what I would like to be able to do in a streaming pipeline.
```golang
func BatchRpcFn {
client RpcClient
bufferRequest *RpcRequest
}

func (f *BatchRpcFn) Setup(ctx context.Context) {
// setup client
}

func (f *BatchRpcFn) ProcessBundle(ctx context.Context, id string, emit func(string, bool)) error {
f.bufferRequest.Ids = append(f.bufferRequest.Ids, id)
if len(f.bufferRequest.Ids) > bufferLimit {
return f.performRequestAndEmit(ctx, emit)
}
return nil
}

func (f *BatchRpcFn) FinishBundle(ctx context.Context, emit func(string, bool)) error {
return f.performRequestAndEmit(ctx, emit)
}
```

### Issue Priority

Priority: 2

### Issue Component

Component: sdk-go

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.