[BEAM-8292] Portable Reshuffle for Go SDK #11197

lostluck · 2020-03-23T17:37:29Z

This adds a Reshuffle transform to the Go SDK.

In particular, it configures windowing & trigggers for a GBK to allow for fusion breaks, where parallelism needs to increase, or decrease due to data bundling properties.
Previous element window and timestamps are preserved.
The SDK operations are wrapped with a higher level reshuffle URN so runners can optimize this step better.
The Go Direct Runner is aware of the Reshuffle and correctly ignores it, as it's a single bundle runner.

Note: While this should work for streaming cases, it hasn't been tested with them yet, due to the current state of streaming the Go SDK.

It further adds one small optimization for the internal Decoding interface, called the DecodeTo method to avoid extra allocations to the heap incurred by returning a *FullValue. Can't avoid extra allocations for KV types at present, but value PCollections should have lower overhead. A subsequent PR will use the DecodeTo method at a other applicable places in the decode stack.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

lostluck · 2020-03-23T17:39:01Z

Run Go Postcommit

lostluck · 2020-03-23T19:20:45Z

R: @youngoli

lostluck · 2020-03-23T19:44:54Z

I'm definitely not merging this until both the PostCommit runs, and someone more familiar with windowing/trigger semantics looks over the configuration I copied over from python:
https://github.com/apache/beam/pull/11197/files#diff-ef420fdb9afbce0674282b4ed4481042R530

lostluck · 2020-03-23T20:23:27Z

Retest this please

lostluck · 2020-03-23T20:55:54Z

Run Go Postcommit

lostluck · 2020-03-23T21:20:10Z

Post commits run and pass which is a good sign!

youngoli · 2020-03-23T23:38:41Z

sdks/go/pkg/beam/core/runtime/exec/coder.go

+		return err
+	}
+	*fv = FullValue{Elm: val}
+	return err


I know this is just preserving the existing behavior, but it seems weird to return err here instead of return nil, even if it is guaranteed to be nil at this point.

Good catch. Thanks!

sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

youngoli · 2020-03-24T01:33:12Z

sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

+
+// FinishBundle propagates finish bundle to downstream nodes.
+func (n *ReshuffleOutput) FinishBundle(ctx context.Context) error {
+	n.b = bytes.Buffer{}


Should this also be clearing n.ret like ReshuffleInput does?

Yes it should. Thanks!

youngoli · 2020-03-24T01:37:28Z

sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

+func (n *ReshuffleOutput) ProcessElement(ctx context.Context, value *FullValue, values ...ReStream) error {
+	// Marshal the pieces into a temporary buffer since they must be transmitted on FnAPI as a single
+	// unit.
+	vs, err := values[0].Open()


It's strange to me that values[] can have multiple elements, but this method ends up actually reading all the values from the first element of it. Could you explain why that happens? Are there sometimes multiple ReStreams representing different things?

This is the Go SDK future proofing itself, against CoGBK supporting multiple datastreams from the runner. Functionally, only datasource.go would need to change in that case. You can see a comment to that effect in datasource.go, and then nearly everything else deals with the values streams properly under that assumption.

In this case, we know that if this code is being used, it's coming from a single GBK, which means there's only a single stream of values, and then since we're framework side, we just handle the stream directly. In that way, it's similar to how we're handling CoGBKs presently, with synthetic inject and expand steps to get to the right number of joined streams, even though the Runner is only providing us with a single data stream for the grouped data.

youngoli · 2020-03-24T03:42:59Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

+	gbk := &pb.PTransform{
+		UniqueName: gbkID,
+		Spec:       &pb.FunctionSpec{Urn: URNGBK},
+		Inputs:     map[string]string{"i0": postReify},


Is this input supposed to be postReify? I would've expected inputID from the previous step.

Yup. postReify is the PCollection that has the random keys and the full windowed value as serialized bytes.
The input from the previous step is used on line 577.
in represents strictly inbound data (and specifically the main input), and From indicates the "Node" in the graph. In this model of the pipeline PCollections are Nodes, and Transforms are Edges.

Oh ok, I see it now. I was completely misinterpreting the expansion here. So if I understand correctly, it should look like this, right? (With pcollections/nodes in square brackets and transforms/edges as arrows)

[in.From] ---input---> [postReify] ---gbk---> [gbkOut] ---output---> [out.To]

Where input and output are the newly added Reshuffle transforms. That looks right to me, and rereading the code it looks consistent with that.

sdks/go/pkg/beam/gbk.go

lukecwik · 2020-03-24T15:06:00Z

R: @reuvenlax

same code, but shorter Co-Authored-By: Daniel Oliveira <younghoono@gmail.com>

lostluck

PTAL

lostluck · 2020-03-24T16:47:43Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

+	gbk := &pb.PTransform{
+		UniqueName: gbkID,
+		Spec:       &pb.FunctionSpec{Urn: URNGBK},
+		Inputs:     map[string]string{"i0": postReify},


Yup. postReify is the PCollection that has the random keys and the full windowed value as serialized bytes.
The input from the previous step is used on line 577.
in represents strictly inbound data (and specifically the main input), and From indicates the "Node" in the graph. In this model of the pipeline PCollections are Nodes, and Transforms are Edges.

lostluck · 2020-03-24T16:51:58Z

sdks/go/pkg/beam/core/runtime/exec/coder.go

+		return err
+	}
+	*fv = FullValue{Elm: val}
+	return err


Good catch. Thanks!

lostluck · 2020-03-24T16:59:48Z

sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

+func (n *ReshuffleOutput) ProcessElement(ctx context.Context, value *FullValue, values ...ReStream) error {
+	// Marshal the pieces into a temporary buffer since they must be transmitted on FnAPI as a single
+	// unit.
+	vs, err := values[0].Open()


This is the Go SDK future proofing itself, against CoGBK supporting multiple datastreams from the runner. Functionally, only datasource.go would need to change in that case. You can see a comment to that effect in datasource.go, and then nearly everything else deals with the values streams properly under that assumption.

In this case, we know that if this code is being used, it's coming from a single GBK, which means there's only a single stream of values, and then since we're framework side, we just handle the stream directly. In that way, it's similar to how we're handling CoGBKs presently, with synthetic inject and expand steps to get to the right number of joined streams, even though the Runner is only providing us with a single data stream for the grouped data.

lostluck · 2020-03-24T17:01:26Z

sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

+
+// FinishBundle propagates finish bundle to downstream nodes.
+func (n *ReshuffleOutput) FinishBundle(ctx context.Context) error {
+	n.b = bytes.Buffer{}


Yes it should. Thanks!

lostluck · 2020-03-24T17:03:24Z

Run Go Postcommit

youngoli

Looks good to me.

dpmills · 2020-03-27T18:41:06Z

sdks/go/pkg/beam/core/runtime/graphx/translate.go

+				WindowFn: makeWindowFn(wfn),
+				// ...output after every element is received...
+				Trigger: &pb.Trigger{
+					// Should this be an Always trigger instead?


In answer to this comment, yes it should be Always

lostluck · 2020-03-27T18:52:17Z

Retest this please

lostluck · 2020-03-27T19:01:20Z

Run Go Postcommit

This adds a Reshuffle transform to the Go SDK. In particular, it configures windowing & trigggers for a GBK to allow for fusion breaks, where parallelism needs to increase, or decrease due to data bundling properties. Previous element window and timestamps are preserved. The SDK operations are wrapped with a higher level reshuffle URN so runners can optimize this step better. The Go Direct Runner is aware of the Reshuffle and correctly ignores it, as it's a single bundle runner. Note: While this should work for streaming cases, it hasn't been tested with them yet, due to the current state of streaming the Go SDK. Co-authored-by: lostluck <13907733+lostluck@users.noreply.github.com> Co-authored-by: Daniel Oliveira <younghoono@gmail.com>

[BEAM-8292] Portable Reshuffle for Go SDK

9fd411d

probot-autolabeler bot added dataflow direct go runners labels Mar 23, 2020

Fix placement of exec translation.

968e45f

youngoli reviewed Mar 24, 2020

View reviewed changes

lukecwik requested a review from reuvenlax March 24, 2020 15:06

lostluck and others added 2 commits March 24, 2020 09:52

Update sdks/go/pkg/beam/core/runtime/exec/reshuffle.go

dbd6d0a

same code, but shorter Co-Authored-By: Daniel Oliveira <younghoono@gmail.com>

review comments

250ea89

lostluck commented Mar 24, 2020

View reviewed changes

youngoli approved these changes Mar 24, 2020

View reviewed changes

dpmills reviewed Mar 27, 2020

View reviewed changes

Switch from ElementCount(1) to Always trigger.

cf16217

lostluck merged commit bbc0c18 into apache:master Mar 27, 2020

lostluck deleted the reshuffle branch March 27, 2020 23:40

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-8292] Portable Reshuffle for Go SDK #11197

[BEAM-8292] Portable Reshuffle for Go SDK #11197

lostluck commented Mar 23, 2020 •

edited

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

youngoli Mar 23, 2020

lostluck Mar 24, 2020

youngoli Mar 24, 2020

lostluck Mar 24, 2020

youngoli Mar 24, 2020

lostluck Mar 24, 2020

youngoli Mar 24, 2020

lostluck Mar 24, 2020

youngoli Mar 24, 2020

lukecwik commented Mar 24, 2020

lostluck left a comment

lostluck Mar 24, 2020

lostluck Mar 24, 2020

lostluck Mar 24, 2020

lostluck Mar 24, 2020

lostluck commented Mar 24, 2020

youngoli left a comment

dpmills Mar 27, 2020

lostluck Mar 27, 2020

lostluck commented Mar 27, 2020

lostluck commented Mar 27, 2020

[BEAM-8292] Portable Reshuffle for Go SDK #11197

[BEAM-8292] Portable Reshuffle for Go SDK #11197

Conversation

lostluck commented Mar 23, 2020 • edited

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

lostluck commented Mar 23, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lukecwik commented Mar 24, 2020

lostluck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lostluck commented Mar 24, 2020

youngoli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lostluck commented Mar 27, 2020

lostluck commented Mar 27, 2020

lostluck commented Mar 23, 2020 •

edited