You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/09/08 18:23:42 UTC
[GitHub] [beam] lostluck opened a new issue, #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
lostluck opened a new issue, #23106:
URL: https://github.com/apache/beam/issues/23106
### What would you like to happen?
Requires implementing a Periodic Impulse and Periodic Sequence transform for occasional re-reads of some data, in a large window.
Requires using those re-reads as a side input.
Requires allowing the map_windows urn: https://github.com/apache/beam/blob/master/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto#L296
So that the SDK can map windows from one to another, this needs to be added to the big switch in exec/translate.go.
Currently, trying to manually use this pattern leads to a panic on execution when a runner makes the mapping request:
https://github.com/apache/beam/blob/master/sdks/go/pkg/beam/core/runtime/exec/translate.go#L734
### Issue Priority
Priority: 1
### Issue Component
Component: sdk-go
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] manuzhang commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
manuzhang commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1241773174
set to p2 since this is a feature request
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] camphillips22 commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
camphillips22 commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1318712021
Thanks for the feedback!
> decoding the appropriate payload proto, and using that
information to configure an exec node that exists on the exec side of the
pipeline, in this case, taking an timestamp, and producing the appropriate
window for it, based on the provided windowing strategy (I assume).
Specific payload (likely a specific proto message) should be listed in the
proto next to the URN enum in beam_fn_api.proto.
I actually got this far myself, but I'm having trouble understanding how to write the node's `ProcessElement`. It's not clear to me what the shape of the data is here and there's not another node that's similar, so it's taking some time to understand how to write that bit. From my experimentation, the `FullValue` is empty, so I'm assuming that I need create the coders and evaluate the `ReStream`, but haven't gotten to that yet.
As far as integration tests, I also took the approach you suggested with a pipeline that looks like:
```golang
func MapWindows(s beam.Scope) {
col := beam.ParDo(s, &makeTimestampedData{Data: []int{4, 9, 2, 3, 5, 7, 8, 1, 6}}, beam.Impulse(s))
windowed := beam.WindowInto(s, window.NewFixedWindows(3*time.Second), col)
mean := stats.Mean(s, windowed)
filtered := beam.ParDo(s, filterAbove, windowed, beam.SideInput{Input: mean})
globalFiltered := beam.WindowInto(s, window.NewGlobalWindows(), filtered)
passert.Sum(s, globalFiltered, "a", 4, 30)
}
```
The problem with this was that it never hit the map_windows URN. Not sure if there's some difference between streaming and batch pipelines here though. I wasn't sure how to test a streaming pipeline in a test like this (well, presumably the `teststream` package, but that's not supported by dataflow), so I ended up slightly modifying the streaming wordcap example to test it on dataflow. Not spending money though would be great, so I'll try to get local flink set up.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern [beam]
Posted by "damccorm (via GitHub) via github" <gi...@beam.apache.org>.
damccorm closed issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
URL: https://github.com/apache/beam/issues/23106
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hnnsgstfssn commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by "hnnsgstfssn (via GitHub)" <gi...@apache.org>.
hnnsgstfssn commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1464889413
I've added a draft of the periodic.Sequence and periodic.Impulse but I'm having some issues and need some guidance around testing.
#25808 includes an example that I'm using to run this on Dataflow.
The pipeline runs and seems to correctly use the side input lookup as it receives input, however when draining it spits out the following trace.
<details>
Error message from worker: generic::unknown: process bundle failed for instruction process_bundle-5-4 using plan drain-S02-11 : panic: runtime error: index out of range [2] with length 2
Full error:
while executing Process for Plan[drain-S02-11]:
2: DataSink[S[ptransform-9@localhost:12371]] Coder:W;coder-50<KV;coder-51<int[varintz;c2];coder-52,string;coder-53>>!IWC
3: PCollection[pcollection-32] Out:[2]
4: WindowInto[FIX[1m0s]]. Out:2
5: PCollection[pcollection-26] Out:[4]
6: ParDo[main.update] Out:[5] Sig: func(context.Context, mtime.Time, int64, func(int, string))
7: PCollection[pcollection-22] Out:[6]
8: SDF.ProcessSizedElementsAndRestrictions[periodic.sequenceGenDoFn] UID:8 Out:[7]
9: PCollection[pcollection-12-truncate-output] Out:[8]
10: SDF.TruncateSizedRestriction[periodic.sequenceGenDoFn] UID:10 Out:[9]
1: DataSource[S[ptransform-8@localhost:12371], 0] Out:10 Coder:W;coder-38<KV;coder-39<KV;coder-40<LP;coder-41<R[periodic.SequenceDefinition]>,KV;coder-42<offsetrange.Restriction[offsetrange.Restriction;c10];coder-43,bool;coder-44>>,double;coder-45>>!GWC
caused by:
panic: runtime error: index out of range [2] with length 2 goroutine 44 [running]:
runtime/debug.Stack()
/usr/lib/go/src/runtime/debug/stack.go:24 +0x65
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.callNoPanic.func1()
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/util.go:58 +0xa5
panic({0xfb70a0, 0xc000137f08})
/usr/lib/go/src/runtime/panic.go:884 +0x213
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*trInvoker).Invoke(0xc0001db0a0?, {0x11f5450?, 0xc00026b800?}, {0xf919c0?, 0xc000011728?}, 0xc0001db0a0?)
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go:320 +0x1b3
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*TruncateSizedRestriction).ProcessElement(0xc00026b7c0, {0x11f5450, 0xc00026b800}, 0xc0001dae00, {0x0, 0x0, 0x0})
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/sdf.go:345 +0x118
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*DataSource).Process(0xc000496c80, {0x11f5450, 0xc00026b800})
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/datasource.go:189 +0x510
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*Plan).Execute.func2({0x11f5450?, 0xc00026b800?})
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/plan.go:131 +0x42
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.callNoPanic({0x11f5450?, 0xc00026b800?}, 0x0?)
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/util.go:62 +0x6c
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/exec.(*Plan).Execute(0xc0000d3c20, {0x11f5450, 0xc00026b800}, {0xc000137ae8, 0x12}, {{0x11ed500?, 0xc0002a0ae0?}, {0x120a9f8?, 0xc0002a0b10?}})
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/exec/plan.go:130 +0x3da
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/harness.(*control).handleInstruction(0xc0004fc000, {0x11f5338, 0xc0004d91a0}, 0xc00009a3c0)
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/harness/harness.go:407 +0xab7
github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/harness.Main.func4({0x11f5338, 0xc0004d91a0}, 0xc00009a3c0)
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/harness/harness.go:193 +0x19d
created by github.com/apache/beam/sdks/v2/go/pkg/beam/core/runtime/harness.Main
/home/rru/temp/beam/sdks/go/pkg/beam/core/runtime/harness/harness.go:212 +0xfed
</details>
I'm now beyond my understanding of the SDK, but it seems it fails [here](https://github.com/apache/beam/blob/cc2a783bc6a722f73bd0ba90b80b719f34377d1b/sdks/go/pkg/beam/core/runtime/exec/sdf_invokers.go#L320) somehow thinking that it's a KV pair that's coming in (?), judging from the caller [here](https://github.com/apache/beam/blob/cc2a783bc6a722f73bd0ba90b80b719f34377d1b/sdks/go/pkg/beam/core/runtime/exec/sdf.go#L326).
I'm not sure what's going on and would appreciate some guidance on addressing this. Perhaps I'm just missing something simple in the implementation?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1317932566
OK, I love that we can do charts in the markdown of an issue. Spectacular!
It's a matter of implementing the URN in the switch statement in
exec/translate.go:, decoding the appropriate payload proto, and using that
information to configure an exec node that exists on the exec side of the
pipeline, in this case, taking an timestamp, and producing the appropriate
window for it, based on the provided windowing strategy (I assume).
Specific payload (likely a specific proto message) should be listed in the
proto next to the URN enum in beam_fn_api.proto.
It's not particularly difficult, just a small pain to test, since I don't
know how well the various semantics are implemented for portable use,
outside of Dataflow. Flink should be able to do it, and I recommend that
for local iteration. Portable Python has issues with streaming semantics.
Go Direct runner, is right out at present (and is in the process of being
replaced with a portable implementation because Direct Runners have
problems.)
So, my recommendation would be to add an integration test to the primitives
package (
https://github.com/apache/beam/blob/master/sdks/go/test/integration/primitives/windowinto.go).
The simple pipeline you have there is a great base and the implementer can
follow the patterns with the existing integration tests, assuming it can be
done in a batch mode. But then it can be run against a few of the runners
at once *and then* the changes implement the changes in the SDK side. The
test can then be filtered out from non-compliant runners in
https://github.com/apache/beam/blob/master/sdks/go/test/integration/integration.go#L63,
as determined by "what fails after it's working well on Dataflow and Flink".
That slows down iteration time a bit (outside of simple unit testing), but
ends up with something that will be tested and work on all the runners,
without spending money on Dataflow, which shouldn't be a pre-req of Beam
Development....
On Wed, 16 Nov 2022 at 12:30, camphillips22 ***@***.***>
wrote:
> How hard would it be to just add the map_windows urn support? I wrote a
> pipeline with a side input that isn't exactly the "slow changing side
> input" pattern, but the pipeline failed because map_windows isn't
> implemented.
>
> For context, the pipeline was shaped approximately like:
>
> flowchart TD
>
> A[WindowInto] --> |"KV#lt;string, data#gt;"| B["Distinct(DropValue(col))"];
> B --> C[Lookup configuration for keys];
> C --> D["Map to KV#lt;string, config#gt;"];
> D --> E
> A --> |"KV#lt;string, data#gt;"| E["Emit new data based on config<br>ParDo(col, beam.SideInput(KV#lt;string, config#gt;)"];
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/beam/issues/23106#issuecomment-1317629945>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ADKDOFKGXFX7TIWZ2QUD25DWIU76PANCNFSM6AAAAAAQIAHRAE>
> .
> You are receiving this because you authored the thread.Message ID:
> ***@***.***>
>
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1319129923
I've been told that one needs to start the local Flink runner in "checkpointing" mode in order to get the desired behavior out of it for streaming things. I don't know how to do that off hand, but I'd look into how the Python Streaming Load Tests set up their flink instances.
I have also confirmed that a Streaming Splittable DoFn (one that returns ProcessContinuations) should be gracefully ending a streaming pipeline when all instances have returned a Stop, so we may have a tractable streaming testing scenario that doesn't require a forced cancellation.
I don't love that we need to have a different configuration for Flink to behave properly in streaming contexts, but that's a different problem.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lostluck commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
lostluck commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1318804162
That's what I was afraid of :/. We don't have much in the way of local
streaming test infra, so flink and Dataflow are the only places to run
things for now. (The problem here is turning down the pipeline at the end.
Beam doesn't support "terminating" a streaming pipeline from the inside so
it needs to happen outside.)
I'll look this up later, but for map windows there should be no "values" it
should basically be just the EventTime timestamp, and the output would be
the IntervalWindow for the configured windowing strategy. The question is
if that's moving as raw data or as a windowed value header with a nonce
value...
On Thu, Nov 17, 2022, 6:22 AM camphillips22 ***@***.***>
wrote:
> Thanks for the feedback!
>
> decoding the appropriate payload proto, and using that
> information to configure an exec node that exists on the exec side of the
> pipeline, in this case, taking an timestamp, and producing the appropriate
> window for it, based on the provided windowing strategy (I assume).
> Specific payload (likely a specific proto message) should be listed in the
> proto next to the URN enum in beam_fn_api.proto.
>
> I actually got this far myself, but I'm having trouble understanding how
> to write the node's ProcessElement. It's not clear to me what the shape
> of the data is here and there's not another node that's similar, so it's
> taking some time to understand how to write that bit. From my
> experimentation, the FullValue is empty, so I'm assuming that I need
> create the coders and evaluate the ReStream, but haven't gotten to that
> yet.
>
> As far as integration tests, I also took the approach you suggested with a
> pipeline that looks like:
>
> func MapWindows(s beam.Scope) {
> col := beam.ParDo(s, &makeTimestampedData{Data: []int{4, 9, 2, 3, 5, 7, 8, 1, 6}}, beam.Impulse(s))
> windowed := beam.WindowInto(s, window.NewFixedWindows(3*time.Second), col)
> mean := stats.Mean(s, windowed)
> filtered := beam.ParDo(s, filterAbove, windowed, beam.SideInput{Input: mean})
> globalFiltered := beam.WindowInto(s, window.NewGlobalWindows(), filtered)
> passert.Sum(s, globalFiltered, "a", 4, 30)
> }
>
> The problem with this was that it never hit the map_windows URN. Not sure
> if there's some difference between streaming and batch pipelines here
> though. I wasn't sure how to test a streaming pipeline in a test like this
> (well, presumably the teststream package, but that's not supported by
> dataflow), so I ended up slightly modifying the streaming wordcap example
> to test it on dataflow. Not spending money though would be great, so I'll
> try to get local flink set up.
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/beam/issues/23106#issuecomment-1318712021>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ADKDOFPV42DRT5DZCHASFVDWIY5RLANCNFSM6AAAAAAQIAHRAE>
> .
> You are receiving this because you authored the thread.Message ID:
> ***@***.***>
>
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] hnnsgstfssn commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by "hnnsgstfssn (via GitHub)" <gi...@apache.org>.
hnnsgstfssn commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1464892470
Actually, I've run the failing pipeline off Beam 2.47.0.dev, and now testing it on Beam 2.45.0 it drains successfully. Draining on 2.46.0 fails. Could it be a regression?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] camphillips22 commented on issue #23106: [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern
Posted by GitBox <gi...@apache.org>.
camphillips22 commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-1317629945
How hard would it be to just add the `map_windows` urn support? I wrote a pipeline with a side input that isn't exactly the "slow changing side input" pattern, but the pipeline failed because map_windows isn't implemented.
For context, the pipeline was shaped approximately like:
```mermaid
flowchart TD
A[WindowInto] --> |"KV#lt;string, data#gt;"| B["Distinct(DropValue(col))"];
B --> C[Lookup configuration for keys];
C --> D["Map to KV#lt;string, config#gt;"];
D --> E
A --> |"KV#lt;string, data#gt;"| E["Emit new data based on config<br>ParDo(col, beam.SideInput(KV#lt;string, config#gt;)"];
```
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Feature Request][Go SDK]: Support Slowly Changing Side Input Pattern [beam]
Posted by "damccorm (via GitHub) via github" <gi...@beam.apache.org>.
damccorm commented on issue #23106:
URL: https://github.com/apache/beam/issues/23106#issuecomment-6748958530
Resolved by #24307 (commit 3951a7d0ad9c2bfc57e830afce43827c7e3236ab), which added `map_windows` support to the Go SDK, and #25808 (commit 16cd2a9872224ecf422e552b028f9c84a3bd87fc), which added `periodic.Sequence` and `periodic.Impulse` transforms along with a slowly updating side input example.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org