[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=411508&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411508 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 27/Mar/20 23:40 Start Date: 27/Mar/20 23:40 Worklog Time Spent: 10m Work Description: lostluck commented on pull request #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 411508) Time Spent: 4h (was: 3h 50m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 4h > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=411290&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411290 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 27/Mar/20 19:01 Start Date: 27/Mar/20 19:01 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-605214792 Run Go Postcommit This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 411290) Time Spent: 3h 50m (was: 3h 40m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 3h 50m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=411278&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411278 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 27/Mar/20 18:52 Start Date: 27/Mar/20 18:52 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-605211109 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 411278) Time Spent: 3h 40m (was: 3.5h) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 3h 40m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=411276&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411276
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 27/Mar/20 18:51
Start Date: 27/Mar/20 18:51
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r399474243
##
File path: sdks/go/pkg/beam/core/runtime/graphx/translate.go
##
@@ -473,6 +479,164 @@ func (m *marshaller) addNode(n *graph.Node) string {
return m.makeNode(id, m.coders.Add(n.Coder), n)
}
+// expandReshuffle translates resharding to a composite reshuffle
+// transform.
+//
+// With proper runner support, the SDK doesn't need to do anything.
+// However, we still need to provide a backup plan in terms of other
+// PTransforms in the event the runner doesn't have a native implementation.
+//
+// In particular, the "backup plan" needs to:
+//
+// * Encode the windowed element, preserving timestamps.
+// * Add random keys to the encoded windowed element []bytes
+// * GroupByKey (in the global window).
+// * Explode the resulting elements list.
+// * Decode the windowed element []bytes.
+//
+// While a simple reshard can be written in user terms, (timestamps and windows
+// are accessible to user functions) there are some framework internal
+// optimizations that can be done if the framework is aware of the reshard,
though
+// ideally this is handled on the runner side.
+//
+// User code is able to write reshards, but it's easier to access
+// the window coders framework side, which is critical for the reshard
+// to function with unbounded inputs.
+func (m *marshaller) expandReshuffle(edge NamedEdge) string {
+ id := edgeID(edge.Edge)
+ var kvCoderID, gbkCoderID string
+ {
+ kv := makeUnionCoder()
+ kvCoderID = m.coders.Add(kv)
+ gbkCoderID = m.coders.Add(coder.NewCoGBK(kv.Components))
+ }
+
+ var subtransforms []string
+
+ in := edge.Edge.Input[0]
+
+ origInput := m.addNode(in.From)
+ // We need to preserve the old windowing/triggering here
+ // for re-instatement after the GBK.
+ preservedWSId := m.pcollections[origInput].GetWindowingStrategyId()
+
+ // Get the windowing strategy from before:
+ postReify := fmt.Sprintf("%v_%v_reifyts", nodeID(in.From), id)
+ m.makeNode(postReify, kvCoderID, in.From)
+
+ // We need to replace postReify's windowing strategy with one
appropriate
+ // for reshuffles.
+ {
+ wfn := window.NewGlobalWindows()
+ m.pcollections[postReify].WindowingStrategyId =
+ m.internWindowingStrategy(&pb.WindowingStrategy{
+ // Not segregated by time...
+ WindowFn: makeWindowFn(wfn),
+ // ...output after every element is received...
+ Trigger: &pb.Trigger{
+ // Should this be an Always trigger
instead?
Review comment:
Thanks!
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 411276)
Time Spent: 3.5h (was: 3h 20m)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 3.5h
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=411268&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-411268
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 27/Mar/20 18:41
Start Date: 27/Mar/20 18:41
Worklog Time Spent: 10m
Work Description: dpmills commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r399468862
##
File path: sdks/go/pkg/beam/core/runtime/graphx/translate.go
##
@@ -473,6 +479,164 @@ func (m *marshaller) addNode(n *graph.Node) string {
return m.makeNode(id, m.coders.Add(n.Coder), n)
}
+// expandReshuffle translates resharding to a composite reshuffle
+// transform.
+//
+// With proper runner support, the SDK doesn't need to do anything.
+// However, we still need to provide a backup plan in terms of other
+// PTransforms in the event the runner doesn't have a native implementation.
+//
+// In particular, the "backup plan" needs to:
+//
+// * Encode the windowed element, preserving timestamps.
+// * Add random keys to the encoded windowed element []bytes
+// * GroupByKey (in the global window).
+// * Explode the resulting elements list.
+// * Decode the windowed element []bytes.
+//
+// While a simple reshard can be written in user terms, (timestamps and windows
+// are accessible to user functions) there are some framework internal
+// optimizations that can be done if the framework is aware of the reshard,
though
+// ideally this is handled on the runner side.
+//
+// User code is able to write reshards, but it's easier to access
+// the window coders framework side, which is critical for the reshard
+// to function with unbounded inputs.
+func (m *marshaller) expandReshuffle(edge NamedEdge) string {
+ id := edgeID(edge.Edge)
+ var kvCoderID, gbkCoderID string
+ {
+ kv := makeUnionCoder()
+ kvCoderID = m.coders.Add(kv)
+ gbkCoderID = m.coders.Add(coder.NewCoGBK(kv.Components))
+ }
+
+ var subtransforms []string
+
+ in := edge.Edge.Input[0]
+
+ origInput := m.addNode(in.From)
+ // We need to preserve the old windowing/triggering here
+ // for re-instatement after the GBK.
+ preservedWSId := m.pcollections[origInput].GetWindowingStrategyId()
+
+ // Get the windowing strategy from before:
+ postReify := fmt.Sprintf("%v_%v_reifyts", nodeID(in.From), id)
+ m.makeNode(postReify, kvCoderID, in.From)
+
+ // We need to replace postReify's windowing strategy with one
appropriate
+ // for reshuffles.
+ {
+ wfn := window.NewGlobalWindows()
+ m.pcollections[postReify].WindowingStrategyId =
+ m.internWindowingStrategy(&pb.WindowingStrategy{
+ // Not segregated by time...
+ WindowFn: makeWindowFn(wfn),
+ // ...output after every element is received...
+ Trigger: &pb.Trigger{
+ // Should this be an Always trigger
instead?
Review comment:
In answer to this comment, yes it should be Always
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 411268)
Time Spent: 3h 20m (was: 3h 10m)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 3h 20m
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=409086&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-409086
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 20:49
Start Date: 24/Mar/20 20:49
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r397451550
##
File path: sdks/go/pkg/beam/core/runtime/graphx/translate.go
##
@@ -473,6 +479,164 @@ func (m *marshaller) addNode(n *graph.Node) string {
return m.makeNode(id, m.coders.Add(n.Coder), n)
}
+// expandReshuffle translates resharding to a composite reshuffle
+// transform.
+//
+// With proper runner support, the SDK doesn't need to do anything.
+// However, we still need to provide a backup plan in terms of other
+// PTransforms in the event the runner doesn't have a native implementation.
+//
+// In particular, the "backup plan" needs to:
+//
+// * Encode the windowed element, preserving timestamps.
+// * Add random keys to the encoded windowed element []bytes
+// * GroupByKey (in the global window).
+// * Explode the resulting elements list.
+// * Decode the windowed element []bytes.
+//
+// While a simple reshard can be written in user terms, (timestamps and windows
+// are accessible to user functions) there are some framework internal
+// optimizations that can be done if the framework is aware of the reshard,
though
+// ideally this is handled on the runner side.
+//
+// User code is able to write reshards, but it's easier to access
+// the window coders framework side, which is critical for the reshard
+// to function with unbounded inputs.
+func (m *marshaller) expandReshuffle(edge NamedEdge) string {
+ id := edgeID(edge.Edge)
+ var kvCoderID, gbkCoderID string
+ {
+ kv := makeUnionCoder()
+ kvCoderID = m.coders.Add(kv)
+ gbkCoderID = m.coders.Add(coder.NewCoGBK(kv.Components))
+ }
+
+ var subtransforms []string
+
+ in := edge.Edge.Input[0]
+
+ origInput := m.addNode(in.From)
+ // We need to preserve the old windowing/triggering here
+ // for re-instatement after the GBK.
+ preservedWSId := m.pcollections[origInput].GetWindowingStrategyId()
+
+ // Get the windowing strategy from before:
+ postReify := fmt.Sprintf("%v_%v_reifyts", nodeID(in.From), id)
+ m.makeNode(postReify, kvCoderID, in.From)
+
+ // We need to replace postReify's windowing strategy with one
appropriate
+ // for reshuffles.
+ {
+ wfn := window.NewGlobalWindows()
+ m.pcollections[postReify].WindowingStrategyId =
+ m.internWindowingStrategy(&pb.WindowingStrategy{
+ // Not segregated by time...
+ WindowFn: makeWindowFn(wfn),
+ // ...output after every element is received...
+ Trigger: &pb.Trigger{
+ // Should this be an Always trigger
instead?
+ Trigger: &pb.Trigger_ElementCount_{
+ ElementCount:
&pb.Trigger_ElementCount{
+ ElementCount: 1,
+ },
+ },
+ },
+ // ...and after outputing, discard the output
elements...
+ AccumulationMode:
pb.AccumulationMode_DISCARDING,
+ // ...and since every pane should have 1
element,
+ // try to preserve the timestamp.
+ OutputTime: pb.OutputTime_EARLIEST_IN_PANE,
+ // Defaults copied from
marshalWindowingStrategy.
+ // TODO(BEAM-3304): migrate to user side
operations once trigger support is in.
+ EnvironmentId: m.addDefaultEnv(),
+ MergeStatus: pb.MergeStatus_NON_MERGING,
+ WindowCoderId:
m.coders.AddWindowCoder(makeWindowCoder(wfn)),
+ ClosingBehavior:
pb.ClosingBehavior_EMIT_IF_NONEMPTY,
+ AllowedLateness: 0,
+ OnTimeBehavior: pb.OnTimeBehavior_FIRE_ALWAYS,
+ })
+ }
+
+ // Inputs (i)
+
+ inputID := fmt.Sprintf("%v_reifyts", id)
+ payload := &pb.ParDoPayload{
+ DoFn: &pb.FunctionSpec{
+ Urn: URNReshuffleInput,
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408927&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408927
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 17:03
Start Date: 24/Mar/20 17:03
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r397305306
##
File path: sdks/go/pkg/beam/core/runtime/graphx/translate.go
##
@@ -473,6 +479,164 @@ func (m *marshaller) addNode(n *graph.Node) string {
return m.makeNode(id, m.coders.Add(n.Coder), n)
}
+// expandReshuffle translates resharding to a composite reshuffle
+// transform.
+//
+// With proper runner support, the SDK doesn't need to do anything.
+// However, we still need to provide a backup plan in terms of other
+// PTransforms in the event the runner doesn't have a native implementation.
+//
+// In particular, the "backup plan" needs to:
+//
+// * Encode the windowed element, preserving timestamps.
+// * Add random keys to the encoded windowed element []bytes
+// * GroupByKey (in the global window).
+// * Explode the resulting elements list.
+// * Decode the windowed element []bytes.
+//
+// While a simple reshard can be written in user terms, (timestamps and windows
+// are accessible to user functions) there are some framework internal
+// optimizations that can be done if the framework is aware of the reshard,
though
+// ideally this is handled on the runner side.
+//
+// User code is able to write reshards, but it's easier to access
+// the window coders framework side, which is critical for the reshard
+// to function with unbounded inputs.
+func (m *marshaller) expandReshuffle(edge NamedEdge) string {
+ id := edgeID(edge.Edge)
+ var kvCoderID, gbkCoderID string
+ {
+ kv := makeUnionCoder()
+ kvCoderID = m.coders.Add(kv)
+ gbkCoderID = m.coders.Add(coder.NewCoGBK(kv.Components))
+ }
+
+ var subtransforms []string
+
+ in := edge.Edge.Input[0]
+
+ origInput := m.addNode(in.From)
+ // We need to preserve the old windowing/triggering here
+ // for re-instatement after the GBK.
+ preservedWSId := m.pcollections[origInput].GetWindowingStrategyId()
+
+ // Get the windowing strategy from before:
+ postReify := fmt.Sprintf("%v_%v_reifyts", nodeID(in.From), id)
+ m.makeNode(postReify, kvCoderID, in.From)
+
+ // We need to replace postReify's windowing strategy with one
appropriate
+ // for reshuffles.
+ {
+ wfn := window.NewGlobalWindows()
+ m.pcollections[postReify].WindowingStrategyId =
+ m.internWindowingStrategy(&pb.WindowingStrategy{
+ // Not segregated by time...
+ WindowFn: makeWindowFn(wfn),
+ // ...output after every element is received...
+ Trigger: &pb.Trigger{
+ // Should this be an Always trigger
instead?
+ Trigger: &pb.Trigger_ElementCount_{
+ ElementCount:
&pb.Trigger_ElementCount{
+ ElementCount: 1,
+ },
+ },
+ },
+ // ...and after outputing, discard the output
elements...
+ AccumulationMode:
pb.AccumulationMode_DISCARDING,
+ // ...and since every pane should have 1
element,
+ // try to preserve the timestamp.
+ OutputTime: pb.OutputTime_EARLIEST_IN_PANE,
+ // Defaults copied from
marshalWindowingStrategy.
+ // TODO(BEAM-3304): migrate to user side
operations once trigger support is in.
+ EnvironmentId: m.addDefaultEnv(),
+ MergeStatus: pb.MergeStatus_NON_MERGING,
+ WindowCoderId:
m.coders.AddWindowCoder(makeWindowCoder(wfn)),
+ ClosingBehavior:
pb.ClosingBehavior_EMIT_IF_NONEMPTY,
+ AllowedLateness: 0,
+ OnTimeBehavior: pb.OnTimeBehavior_FIRE_ALWAYS,
+ })
+ }
+
+ // Inputs (i)
+
+ inputID := fmt.Sprintf("%v_reifyts", id)
+ payload := &pb.ParDoPayload{
+ DoFn: &pb.FunctionSpec{
+ Urn: URNReshuffleInput,
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408926&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408926
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 17:03
Start Date: 24/Mar/20 17:03
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r397308582
##
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##
@@ -259,59 +298,82 @@ type customDecoder struct {
dec Decoder
}
-func (c *customDecoder) Decode(r io.Reader) (*FullValue, error) {
+func (c *customDecoder) DecodeTo(r io.Reader, fv *FullValue) error {
// (1) Read length-prefixed encoded data
size, err := coder.DecodeVarInt(r)
if err != nil {
- return nil, err
+ return err
}
data, err := ioutilx.ReadN(r, (int)(size))
if err != nil {
- return nil, err
+ return err
}
// (2) Call decode
val, err := c.dec.Decode(c.t, data)
if err != nil {
+ return err
+ }
+ *fv = FullValue{Elm: val}
+ return err
Review comment:
Good catch. Thanks!
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408926)
Time Spent: 2.5h (was: 2h 20m)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 2.5h
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408928&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408928
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 17:03
Start Date: 24/Mar/20 17:03
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r397314344
##
File path: sdks/go/pkg/beam/core/runtime/exec/reshuffle.go
##
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "math/rand"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
+)
+
+// ReshuffleInput is a Node.
+type ReshuffleInput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the input PCollection.
+ Seed int64
+ Out Node
+
+ r*rand.Rand
+ enc ElementEncoder
+ wEnc WindowEncoder
+ bbytes.Buffer
+ // ret is a cached allocations for passing to the next Unit. Units
never modify the passed in FullValue.
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleInput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleInput) Up(ctx context.Context) error {
+ n.enc = MakeElementEncoder(coder.SkipW(n.Coder))
+ n.wEnc = MakeWindowEncoder(n.Coder.Window)
+ n.r = rand.New(rand.NewSource(n.Seed))
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleInput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+func (n *ReshuffleInput) ProcessElement(ctx context.Context, value *FullValue,
values ...ReStream) error {
+ n.b.Reset()
+ if err := EncodeWindowedValueHeader(n.wEnc, value.Windows,
value.Timestamp, &n.b); err != nil {
+ return err
+ }
+ if err := n.enc.Encode(value, &n.b); err != nil {
+ return errors.WithContextf(err, "encoding element %v with coder
%v", value, n.Coder)
+ }
+ n.ret = FullValue{Elm: n.r.Int(), Elm2: n.b.Bytes(), Timestamp:
value.Timestamp}
+ if err := n.Out.ProcessElement(ctx, &n.ret); err != nil {
+ return err
+ }
+ return nil
+}
+
+// FinishBundle propagates finish bundle, and clears cached state.
+func (n *ReshuffleInput) FinishBundle(ctx context.Context) error {
+ n.b = bytes.Buffer{}
+ n.ret = FullValue{}
+ return MultiFinishBundle(ctx, n.Out)
+}
+
+// Down is a no-op.
+func (n *ReshuffleInput) Down(ctx context.Context) error {
+ return nil
+}
+
+func (n *ReshuffleInput) String() string {
+ return fmt.Sprintf("ReshuffleInput[%v] Coder:%v", n.SID, n.Coder)
+}
+
+// ReshuffleOutput is a Node.
+type ReshuffleOutput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the receiving PCollection.
+ Out Node
+
+ bbytes.Buffer
+ dec ElementDecoder
+ wDec WindowDecoder
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleOutput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleOutput) Up(ctx context.Context) error {
+ n.dec = MakeElementDecoder(coder.SkipW(n.Coder))
+ n.wDec = MakeWindowDecoder(n.Coder.Window)
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleOutput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+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
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408930&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408930 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 24/Mar/20 17:03 Start Date: 24/Mar/20 17:03 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-603372518 Run Go Postcommit This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408930) Time Spent: 3h (was: 2h 50m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 3h > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408929&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408929
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 17:03
Start Date: 24/Mar/20 17:03
Worklog Time Spent: 10m
Work Description: lostluck commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r397315541
##
File path: sdks/go/pkg/beam/core/runtime/exec/reshuffle.go
##
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "math/rand"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
+)
+
+// ReshuffleInput is a Node.
+type ReshuffleInput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the input PCollection.
+ Seed int64
+ Out Node
+
+ r*rand.Rand
+ enc ElementEncoder
+ wEnc WindowEncoder
+ bbytes.Buffer
+ // ret is a cached allocations for passing to the next Unit. Units
never modify the passed in FullValue.
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleInput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleInput) Up(ctx context.Context) error {
+ n.enc = MakeElementEncoder(coder.SkipW(n.Coder))
+ n.wEnc = MakeWindowEncoder(n.Coder.Window)
+ n.r = rand.New(rand.NewSource(n.Seed))
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleInput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+func (n *ReshuffleInput) ProcessElement(ctx context.Context, value *FullValue,
values ...ReStream) error {
+ n.b.Reset()
+ if err := EncodeWindowedValueHeader(n.wEnc, value.Windows,
value.Timestamp, &n.b); err != nil {
+ return err
+ }
+ if err := n.enc.Encode(value, &n.b); err != nil {
+ return errors.WithContextf(err, "encoding element %v with coder
%v", value, n.Coder)
+ }
+ n.ret = FullValue{Elm: n.r.Int(), Elm2: n.b.Bytes(), Timestamp:
value.Timestamp}
+ if err := n.Out.ProcessElement(ctx, &n.ret); err != nil {
+ return err
+ }
+ return nil
+}
+
+// FinishBundle propagates finish bundle, and clears cached state.
+func (n *ReshuffleInput) FinishBundle(ctx context.Context) error {
+ n.b = bytes.Buffer{}
+ n.ret = FullValue{}
+ return MultiFinishBundle(ctx, n.Out)
+}
+
+// Down is a no-op.
+func (n *ReshuffleInput) Down(ctx context.Context) error {
+ return nil
+}
+
+func (n *ReshuffleInput) String() string {
+ return fmt.Sprintf("ReshuffleInput[%v] Coder:%v", n.SID, n.Coder)
+}
+
+// ReshuffleOutput is a Node.
+type ReshuffleOutput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the receiving PCollection.
+ Out Node
+
+ bbytes.Buffer
+ dec ElementDecoder
+ wDec WindowDecoder
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleOutput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleOutput) Up(ctx context.Context) error {
+ n.dec = MakeElementDecoder(coder.SkipW(n.Coder))
+ n.wDec = MakeWindowDecoder(n.Coder.Window)
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleOutput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+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
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408837&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408837 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 24/Mar/20 15:06 Start Date: 24/Mar/20 15:06 Worklog Time Spent: 10m Work Description: lukecwik commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-603292995 R: @reuvenlax This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408837) Time Spent: 2h 20m (was: 2h 10m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 2h 20m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408533&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408533
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396886566
##
File path: sdks/go/pkg/beam/core/runtime/graphx/translate.go
##
@@ -473,6 +479,164 @@ func (m *marshaller) addNode(n *graph.Node) string {
return m.makeNode(id, m.coders.Add(n.Coder), n)
}
+// expandReshuffle translates resharding to a composite reshuffle
+// transform.
+//
+// With proper runner support, the SDK doesn't need to do anything.
+// However, we still need to provide a backup plan in terms of other
+// PTransforms in the event the runner doesn't have a native implementation.
+//
+// In particular, the "backup plan" needs to:
+//
+// * Encode the windowed element, preserving timestamps.
+// * Add random keys to the encoded windowed element []bytes
+// * GroupByKey (in the global window).
+// * Explode the resulting elements list.
+// * Decode the windowed element []bytes.
+//
+// While a simple reshard can be written in user terms, (timestamps and windows
+// are accessible to user functions) there are some framework internal
+// optimizations that can be done if the framework is aware of the reshard,
though
+// ideally this is handled on the runner side.
+//
+// User code is able to write reshards, but it's easier to access
+// the window coders framework side, which is critical for the reshard
+// to function with unbounded inputs.
+func (m *marshaller) expandReshuffle(edge NamedEdge) string {
+ id := edgeID(edge.Edge)
+ var kvCoderID, gbkCoderID string
+ {
+ kv := makeUnionCoder()
+ kvCoderID = m.coders.Add(kv)
+ gbkCoderID = m.coders.Add(coder.NewCoGBK(kv.Components))
+ }
+
+ var subtransforms []string
+
+ in := edge.Edge.Input[0]
+
+ origInput := m.addNode(in.From)
+ // We need to preserve the old windowing/triggering here
+ // for re-instatement after the GBK.
+ preservedWSId := m.pcollections[origInput].GetWindowingStrategyId()
+
+ // Get the windowing strategy from before:
+ postReify := fmt.Sprintf("%v_%v_reifyts", nodeID(in.From), id)
+ m.makeNode(postReify, kvCoderID, in.From)
+
+ // We need to replace postReify's windowing strategy with one
appropriate
+ // for reshuffles.
+ {
+ wfn := window.NewGlobalWindows()
+ m.pcollections[postReify].WindowingStrategyId =
+ m.internWindowingStrategy(&pb.WindowingStrategy{
+ // Not segregated by time...
+ WindowFn: makeWindowFn(wfn),
+ // ...output after every element is received...
+ Trigger: &pb.Trigger{
+ // Should this be an Always trigger
instead?
+ Trigger: &pb.Trigger_ElementCount_{
+ ElementCount:
&pb.Trigger_ElementCount{
+ ElementCount: 1,
+ },
+ },
+ },
+ // ...and after outputing, discard the output
elements...
+ AccumulationMode:
pb.AccumulationMode_DISCARDING,
+ // ...and since every pane should have 1
element,
+ // try to preserve the timestamp.
+ OutputTime: pb.OutputTime_EARLIEST_IN_PANE,
+ // Defaults copied from
marshalWindowingStrategy.
+ // TODO(BEAM-3304): migrate to user side
operations once trigger support is in.
+ EnvironmentId: m.addDefaultEnv(),
+ MergeStatus: pb.MergeStatus_NON_MERGING,
+ WindowCoderId:
m.coders.AddWindowCoder(makeWindowCoder(wfn)),
+ ClosingBehavior:
pb.ClosingBehavior_EMIT_IF_NONEMPTY,
+ AllowedLateness: 0,
+ OnTimeBehavior: pb.OnTimeBehavior_FIRE_ALWAYS,
+ })
+ }
+
+ // Inputs (i)
+
+ inputID := fmt.Sprintf("%v_reifyts", id)
+ payload := &pb.ParDoPayload{
+ DoFn: &pb.FunctionSpec{
+ Urn: URNReshuffleInput,
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408534&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408534
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396856077
##
File path: sdks/go/pkg/beam/core/runtime/exec/reshuffle.go
##
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "math/rand"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
+)
+
+// ReshuffleInput is a Node.
+type ReshuffleInput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the input PCollection.
+ Seed int64
+ Out Node
+
+ r*rand.Rand
+ enc ElementEncoder
+ wEnc WindowEncoder
+ bbytes.Buffer
+ // ret is a cached allocations for passing to the next Unit. Units
never modify the passed in FullValue.
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleInput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleInput) Up(ctx context.Context) error {
+ n.enc = MakeElementEncoder(coder.SkipW(n.Coder))
+ n.wEnc = MakeWindowEncoder(n.Coder.Window)
+ n.r = rand.New(rand.NewSource(n.Seed))
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleInput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+func (n *ReshuffleInput) ProcessElement(ctx context.Context, value *FullValue,
values ...ReStream) error {
+ n.b.Reset()
+ if err := EncodeWindowedValueHeader(n.wEnc, value.Windows,
value.Timestamp, &n.b); err != nil {
+ return err
+ }
+ if err := n.enc.Encode(value, &n.b); err != nil {
+ return errors.WithContextf(err, "encoding element %v with coder
%v", value, n.Coder)
+ }
+ n.ret = FullValue{Elm: n.r.Int(), Elm2: n.b.Bytes(), Timestamp:
value.Timestamp}
+ if err := n.Out.ProcessElement(ctx, &n.ret); err != nil {
+ return err
+ }
+ return nil
+}
+
+// FinishBundle propagates finish bundle, and clears cached state.
+func (n *ReshuffleInput) FinishBundle(ctx context.Context) error {
+ n.b = bytes.Buffer{}
+ n.ret = FullValue{}
+ return MultiFinishBundle(ctx, n.Out)
+}
+
+// Down is a no-op.
+func (n *ReshuffleInput) Down(ctx context.Context) error {
+ return nil
+}
+
+func (n *ReshuffleInput) String() string {
+ return fmt.Sprintf("ReshuffleInput[%v] Coder:%v", n.SID, n.Coder)
+}
+
+// ReshuffleOutput is a Node.
+type ReshuffleOutput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the receiving PCollection.
+ Out Node
+
+ bbytes.Buffer
+ dec ElementDecoder
+ wDec WindowDecoder
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleOutput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleOutput) Up(ctx context.Context) error {
+ n.dec = MakeElementDecoder(coder.SkipW(n.Coder))
+ n.wDec = MakeWindowDecoder(n.Coder.Window)
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleOutput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+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
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408535&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408535
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396887966
##
File path: sdks/go/pkg/beam/gbk.go
##
@@ -95,3 +95,52 @@ func TryCoGroupByKey(s Scope, cols ...PCollection)
(PCollection, error) {
ret.SetCoder(NewCoder(ret.Type()))
return ret, nil
}
+
+// Reshuffle copies a PCollection of the same kind and using the same element
+// coder, and maintains the same windowing information. Importantly, it allows
+// the result PCollection to be processed with a different sharding, in a
+// different stage than the input PCollection.
+//
+// For example, if a computation needs a lot of parallelism but
+// produces only a small amount of output data, then the computation
+// producing the data can run with as much parallelism as needed,
+// while the output file is written with a smaller amount of
+// parallelism, using the following pattern:
+//
+// pc := bigHairyComputationNeedingParallelism(scope) // PCollection
+// resharded := beam.Reshard(scope, pc)// PCollection
+//
+// Another use case is when one has a non-deterministic DoFn followed by one
+// that performs externally-visible side effects. Inserting a Reshard
+// between these DoFns ensures that retries of the second DoFn will always be
+// the same, which is necessary to make side effects idempotent.
+//
+// A Reshuffle will force a break in the optimized pipeline. Consequently,
+// this operation should be used sparingly, only after determining that the
+// pipeline without reshard is broken in some way and performing an extra
+// operation is worth the cost.
+func Reshuffle(s Scope, col PCollection) PCollection {
+ return Must(TryReshuffle(s, col))
+}
+
+// TryReshuffle inserts a Reshard into the pipeline, and returns an error if
Review comment:
Same as previous comment, using Reshard instead of Reshuffle. The error
message a few lines below also does that.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408535)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 2h
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408531&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408531
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396855047
##
File path: sdks/go/pkg/beam/core/runtime/exec/reshuffle.go
##
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "math/rand"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
+)
+
+// ReshuffleInput is a Node.
+type ReshuffleInput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the input PCollection.
+ Seed int64
+ Out Node
+
+ r*rand.Rand
+ enc ElementEncoder
+ wEnc WindowEncoder
+ bbytes.Buffer
+ // ret is a cached allocations for passing to the next Unit. Units
never modify the passed in FullValue.
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleInput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleInput) Up(ctx context.Context) error {
+ n.enc = MakeElementEncoder(coder.SkipW(n.Coder))
+ n.wEnc = MakeWindowEncoder(n.Coder.Window)
+ n.r = rand.New(rand.NewSource(n.Seed))
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleInput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+func (n *ReshuffleInput) ProcessElement(ctx context.Context, value *FullValue,
values ...ReStream) error {
+ n.b.Reset()
+ if err := EncodeWindowedValueHeader(n.wEnc, value.Windows,
value.Timestamp, &n.b); err != nil {
+ return err
+ }
+ if err := n.enc.Encode(value, &n.b); err != nil {
+ return errors.WithContextf(err, "encoding element %v with coder
%v", value, n.Coder)
+ }
+ n.ret = FullValue{Elm: n.r.Int(), Elm2: n.b.Bytes(), Timestamp:
value.Timestamp}
+ if err := n.Out.ProcessElement(ctx, &n.ret); err != nil {
+ return err
+ }
+ return nil
+}
+
+// FinishBundle propagates finish bundle, and clears cached state.
+func (n *ReshuffleInput) FinishBundle(ctx context.Context) error {
+ n.b = bytes.Buffer{}
+ n.ret = FullValue{}
+ return MultiFinishBundle(ctx, n.Out)
+}
+
+// Down is a no-op.
+func (n *ReshuffleInput) Down(ctx context.Context) error {
+ return nil
+}
+
+func (n *ReshuffleInput) String() string {
+ return fmt.Sprintf("ReshuffleInput[%v] Coder:%v", n.SID, n.Coder)
+}
+
+// ReshuffleOutput is a Node.
+type ReshuffleOutput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the receiving PCollection.
+ Out Node
+
+ bbytes.Buffer
+ dec ElementDecoder
+ wDec WindowDecoder
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleOutput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleOutput) Up(ctx context.Context) error {
+ n.dec = MakeElementDecoder(coder.SkipW(n.Coder))
+ n.wDec = MakeWindowDecoder(n.Coder.Window)
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleOutput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+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
+
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408532&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408532
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396853348
##
File path: sdks/go/pkg/beam/core/runtime/exec/reshuffle.go
##
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exec
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "io"
+ "math/rand"
+
+ "github.com/apache/beam/sdks/go/pkg/beam/core/graph/coder"
+ "github.com/apache/beam/sdks/go/pkg/beam/internal/errors"
+)
+
+// ReshuffleInput is a Node.
+type ReshuffleInput struct {
+ UID UnitID
+ SID StreamID
+ Coder *coder.Coder // Coder for the input PCollection.
+ Seed int64
+ Out Node
+
+ r*rand.Rand
+ enc ElementEncoder
+ wEnc WindowEncoder
+ bbytes.Buffer
+ // ret is a cached allocations for passing to the next Unit. Units
never modify the passed in FullValue.
+ ret FullValue
+}
+
+// ID returns the unit debug id.
+func (n *ReshuffleInput) ID() UnitID {
+ return n.UID
+}
+
+// Up initializes the value and window encoders, and the random source.
+func (n *ReshuffleInput) Up(ctx context.Context) error {
+ n.enc = MakeElementEncoder(coder.SkipW(n.Coder))
+ n.wEnc = MakeWindowEncoder(n.Coder.Window)
+ n.r = rand.New(rand.NewSource(n.Seed))
+ return nil
+}
+
+// StartBundle is a no-op.
+func (n *ReshuffleInput) StartBundle(ctx context.Context, id string, data
DataContext) error {
+ return MultiStartBundle(ctx, id, data, n.Out)
+}
+
+func (n *ReshuffleInput) ProcessElement(ctx context.Context, value *FullValue,
values ...ReStream) error {
+ n.b.Reset()
+ if err := EncodeWindowedValueHeader(n.wEnc, value.Windows,
value.Timestamp, &n.b); err != nil {
+ return err
+ }
+ if err := n.enc.Encode(value, &n.b); err != nil {
+ return errors.WithContextf(err, "encoding element %v with coder
%v", value, n.Coder)
+ }
+ n.ret = FullValue{Elm: n.r.Int(), Elm2: n.b.Bytes(), Timestamp:
value.Timestamp}
+ if err := n.Out.ProcessElement(ctx, &n.ret); err != nil {
+ return err
+ }
+ return nil
Review comment:
```suggestion
return n.Out.ProcessElement(ctx, &n.ret)
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408532)
Time Spent: 1h 50m (was: 1h 40m)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 1h 50m
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408536&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408536
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396887810
##
File path: sdks/go/pkg/beam/gbk.go
##
@@ -95,3 +95,52 @@ func TryCoGroupByKey(s Scope, cols ...PCollection)
(PCollection, error) {
ret.SetCoder(NewCoder(ret.Type()))
return ret, nil
}
+
+// Reshuffle copies a PCollection of the same kind and using the same element
+// coder, and maintains the same windowing information. Importantly, it allows
+// the result PCollection to be processed with a different sharding, in a
+// different stage than the input PCollection.
+//
+// For example, if a computation needs a lot of parallelism but
+// produces only a small amount of output data, then the computation
+// producing the data can run with as much parallelism as needed,
+// while the output file is written with a smaller amount of
+// parallelism, using the following pattern:
+//
+// pc := bigHairyComputationNeedingParallelism(scope) // PCollection
+// resharded := beam.Reshard(scope, pc)// PCollection
Review comment:
Here and elsewhere in this comment, Reshuffle is referred to as "Reshard". I
think it's fine to refer to it as a reshard informally, since that's what it
functionally is, but the places where it's used as a proper noun should be
switched to Reshuffle.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408536)
Time Spent: 2h 10m (was: 2h)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 2h 10m
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[
https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408530&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408530
]
ASF GitHub Bot logged work on BEAM-8292:
Author: ASF GitHub Bot
Created on: 24/Mar/20 04:20
Start Date: 24/Mar/20 04:20
Worklog Time Spent: 10m
Work Description: youngoli commented on pull request #11197: [BEAM-8292]
Portable Reshuffle for Go SDK
URL: https://github.com/apache/beam/pull/11197#discussion_r396823199
##
File path: sdks/go/pkg/beam/core/runtime/exec/coder.go
##
@@ -259,59 +298,82 @@ type customDecoder struct {
dec Decoder
}
-func (c *customDecoder) Decode(r io.Reader) (*FullValue, error) {
+func (c *customDecoder) DecodeTo(r io.Reader, fv *FullValue) error {
// (1) Read length-prefixed encoded data
size, err := coder.DecodeVarInt(r)
if err != nil {
- return nil, err
+ return err
}
data, err := ioutilx.ReadN(r, (int)(size))
if err != nil {
- return nil, err
+ return err
}
// (2) Call decode
val, err := c.dec.Decode(c.t, data)
if err != nil {
+ return err
+ }
+ *fv = FullValue{Elm: val}
+ return err
Review comment:
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.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 408530)
Time Spent: 1h 40m (was: 1.5h)
> Add a Reshuffle PTransform preventing fusion of the surrounding transforms
> --
>
> Key: BEAM-8292
> URL: https://issues.apache.org/jira/browse/BEAM-8292
> Project: Beam
> Issue Type: New Feature
> Components: sdk-go
>Reporter: John Patoch
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 1h 40m
> Remaining Estimate: 0h
>
> Reshuffle is a PTransform that takes a PCollection and shuffles the data
> to help increase parallelism.
> Reshuffle adds a temporary random key to each element, performs a
> GroupByKey, and finally removes the temporary key.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408326&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408326 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 21:20 Start Date: 23/Mar/20 21:20 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602862121 Post commits run and pass which is a good sign! This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408326) Time Spent: 1.5h (was: 1h 20m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 1.5h > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408303&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408303 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 20:55 Start Date: 23/Mar/20 20:55 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602851450 Run Go Postcommit This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408303) Time Spent: 1h 20m (was: 1h 10m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 1h 20m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408251&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408251 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 20:25 Start Date: 23/Mar/20 20:25 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602835398 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408251) Time Spent: 1h 10m (was: 1h) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 1h 10m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408210&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408210 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 19:44 Start Date: 23/Mar/20 19:44 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602817069 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 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408210) Time Spent: 1h (was: 50m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 1h > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408187&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408187 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 19:20 Start Date: 23/Mar/20 19:20 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602805178 R: @youngoli This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408187) Time Spent: 50m (was: 40m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 50m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408093&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408093 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 17:39 Start Date: 23/Mar/20 17:39 Worklog Time Spent: 10m Work Description: lostluck commented on issue #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197#issuecomment-602750930 Run Go Postcommit This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 408093) Time Spent: 40m (was: 0.5h) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Assignee: Robert Burke >Priority: Minor > Time Spent: 40m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=408091&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-408091 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 23/Mar/20 17:37 Start Date: 23/Mar/20 17:37 Worklog Time Spent: 10m Work Description: lostluck commented on pull request #11197: [BEAM-8292] Portable Reshuffle for Go SDK URL: https://github.com/apache/beam/pull/11197 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. 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)**](https://beam.apache.org/contribute/#make-your-change) 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](https://www.apache.org/licenses/icla.pdf). See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier). Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastComp
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=316325&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-316325 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 22/Sep/19 20:10 Start Date: 22/Sep/19 20:10 Worklog Time Spent: 10m Work Description: angulartist commented on pull request #9634: [BEAM-8292] Add a Reshuffle PTransform preventing fusion of the surrounding transforms URL: https://github.com/apache/beam/pull/9634 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 316325) Time Spent: 20m (was: 10m) > Add a Reshuffle PTransform preventing fusion of the surrounding transforms > -- > > Key: BEAM-8292 > URL: https://issues.apache.org/jira/browse/BEAM-8292 > Project: Beam > Issue Type: New Feature > Components: sdk-go >Reporter: John Patoch >Priority: Minor > Time Spent: 20m > Remaining Estimate: 0h > > Reshuffle is a PTransform that takes a PCollection and shuffles the data > to help increase parallelism. > Reshuffle adds a temporary random key to each element, performs a > GroupByKey, and finally removes the temporary key. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-8292) Add a Reshuffle PTransform preventing fusion of the surrounding transforms
[ https://issues.apache.org/jira/browse/BEAM-8292?focusedWorklogId=316323&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-316323 ] ASF GitHub Bot logged work on BEAM-8292: Author: ASF GitHub Bot Created on: 22/Sep/19 20:09 Start Date: 22/Sep/19 20:09 Worklog Time Spent: 10m Work Description: angulartist commented on pull request #9634: [BEAM-8292] Add a Reshuffle PTransform preventing fusion of the surrounding transforms URL: https://github.com/apache/beam/pull/9634 Reshuffle is a PTransform that takes a PCollection and shuffles the data to help increase parallelism. Reshuffle adds a temporary random key to each element, performs a GroupByKey, and finally removes the temporary key. Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) 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. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf). Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/) Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)[](https://bui
