hnnsgstfssn commented on code in PR #25808: URL: https://github.com/apache/beam/pull/25808#discussion_r1133221543
########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) + return offsetrange.Restriction{ + Start: int64(0), + End: int64(totalOutputs), + } +} + +func (fn *sequenceGenDoFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker { + return sdf.NewLockRTracker(offsetrange.NewTracker(rest)) +} + +func (fn *sequenceGenDoFn) RestrictionSize(_ SequenceDefinition, rest offsetrange.Restriction) float64 { + return rest.Size() +} + +func (fn *sequenceGenDoFn) SplitRestriction(_ SequenceDefinition, rest offsetrange.Restriction) []offsetrange.Restriction { + return []offsetrange.Restriction{rest} +} + +// TruncateRestriction immediately truncates the entire restrication. +func (fn *sequenceGenDoFn) TruncateRestriction(_ *sdf.LockRTracker, _ SequenceDefinition) offsetrange.Restriction { + return offsetrange.Restriction{} +} + +func (fn *sequenceGenDoFn) CreateWatermarkEstimator() *sdf.ManualWatermarkEstimator { + return &sdf.ManualWatermarkEstimator{} Review Comment: Was thinking, while this zero `time.Time` value seems to work, it might be clearer if it was initialized to `mtime.MinTimestamp.ToTime()`. What do you think? ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) Review Comment: To satisfy the linter I've changed this to `mtime.Time(sd.End).ToTime().Sub(mtime.Time(sd.Start).ToTime()) / sd.Interval`. ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time Review Comment: Indeed I struggled with the direct runner, but have now added two working tests that uses the `prism` runner. I am struggling to understand how to fold this into the sequence definition as you suggest, to properly test it. Instead I went ahead and removed this entirely and the DoFn uses `time.Now`. I am happy to leave it there for now. If you want to expand on how to fold it into the definition and test it I am happy to add this as well. ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) + return offsetrange.Restriction{ + Start: int64(0), + End: int64(totalOutputs), + } +} + +func (fn *sequenceGenDoFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker { + return sdf.NewLockRTracker(offsetrange.NewTracker(rest)) +} + +func (fn *sequenceGenDoFn) RestrictionSize(_ SequenceDefinition, rest offsetrange.Restriction) float64 { + return rest.Size() +} + +func (fn *sequenceGenDoFn) SplitRestriction(_ SequenceDefinition, rest offsetrange.Restriction) []offsetrange.Restriction { + return []offsetrange.Restriction{rest} +} + +// TruncateRestriction immediately truncates the entire restrication. +func (fn *sequenceGenDoFn) TruncateRestriction(_ *sdf.LockRTracker, _ SequenceDefinition) offsetrange.Restriction { + return offsetrange.Restriction{} +} + +func (fn *sequenceGenDoFn) CreateWatermarkEstimator() *sdf.ManualWatermarkEstimator { + return &sdf.ManualWatermarkEstimator{} +} + +func (fn *sequenceGenDoFn) ProcessElement(ctx context.Context, we *sdf.ManualWatermarkEstimator, rt *sdf.LockRTracker, sd SequenceDefinition, emit func(beam.EventTime, int64)) (sdf.ProcessContinuation, error) { + currentOutputIndex := rt.GetRestriction().(offsetrange.Restriction).Start + currentOutputTimestamp := sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime := fn.now() + we.UpdateWatermark(currentOutputTimestamp) + for currentOutputTimestamp.Before(currentTime) { + if rt.TryClaim(currentOutputIndex) { + emit(mtime.FromTime(currentOutputTimestamp), currentOutputTimestamp.UnixMilli()) + currentOutputIndex += 1 + currentOutputTimestamp = sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime = fn.now() + we.UpdateWatermark(currentOutputTimestamp) + } else if err := rt.GetError(); err != nil || rt.IsDone() { + // Stop processing on error or completion + return sdf.StopProcessing(), rt.GetError() + } else { + return sdf.ResumeProcessingIn(sd.Interval), nil + } + } + + return sdf.ResumeProcessingIn(time.Until(currentOutputTimestamp)), nil +} + +type impulseConfig struct { + ApplyWindow bool + + now func() time.Time +} + +type impulseOption func(*impulseConfig) error + +// ImpulseOption is a function that configures an [Impulse] transform. +type ImpulseOption = impulseOption + +// WithApplyWindow configures the [Impulse] transform to apply a fixed window +// transform to the output PCollection. +func WithApplyWindow() ImpulseOption { + return func(o *impulseConfig) error { + o.ApplyWindow = true + return nil + } +} + +func withNowFunc(now func() time.Time) ImpulseOption { + return func(o *impulseConfig) error { + o.now = now + return nil + } +} + +// Impulse is a PTransform which generates a sequence of timestamped +// elements at fixed runtime intervals. If [WithApplyWindow] is specified, each +// element will be assigned to its own fixed window of interval size. +// +// The transform behaves the same as [Sequence] transform, but can be +// used as the first transform in a pipeline. +// +// The following applies to the arguments. +// - if interval <= 0, interval is set to [math.MaxInt64] +// - if start is a zero value [time.Time], start is set to the current time +// - if start is after end, start is set to end +// +// The PCollection generated by Impulse is unbounded and the output elements +// are the [time.UnixMilli] int64 values of the output timestamp. +func Impulse(s beam.Scope, start, end time.Time, interval time.Duration, opts ...ImpulseOption) beam.PCollection { + if interval <= 0 { + interval = math.MaxInt64 + } + if start.IsZero() { + start = time.Now() + } + if start.After(end) { + start = end + } Review Comment: Thanks for sharing your thoughts here. (1) Since we are not able to use `mtime.Time` directly, I agree we should instead use `int64`s to store the `time.UnixMilli` values on `SequenceDefinition`. I have made this change. (2) I agree that it is odd to send the duplicate `int64` values and that emitting `[]byte` better aligns with the existing `beam.Impulse`. I also agree that it makes sense to output `int64` from the `Sequence` transform. I have changed `sequenceGenDoFn` to output `[]byte`, but the `Sequence` transform still shares the same implementation and will also output `[]byte`. Can you suggest a neat way to share the implementation but still return `int64` from the `Sequence` transform? (3) I've slightly reworked the validation of start, end and interval. Start and end are normalized into `[mtime.MinTimestamp,mtime.MaxTimestamp]` using `mtime.Normalize` and the interval is set to `end.Sub(start)` if it is unset or invalid. I wasn't sure which way to go based on your feedback, but let me know what you think about this change. ########## sdks/go/examples/slowly_updating_side_input/slowly_updating_side_input.go: ########## @@ -0,0 +1,145 @@ +// 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 main + +import ( + "context" + "flag" + "strings" + "time" + + "cloud.google.com/go/pubsub" + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio" + "github.com/apache/beam/sdks/v2/go/pkg/beam/log" + "github.com/apache/beam/sdks/v2/go/pkg/beam/options/gcpopts" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" + _ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/dataflow" + "github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/periodic" + "github.com/apache/beam/sdks/v2/go/pkg/beam/util/pubsubx" +) + +func init() { + register.Function4x0(update) + register.Function4x0(process) + register.Emitter2[int, string]() + register.Iter1[string]() +} + +// update simulates an external call to get data for the side input. +func update(ctx context.Context, t beam.EventTime, i int64, emit func(int, string)) { + log.Infof(ctx, "Making external call %d at %s", i, t.ToTime().Format(time.RFC3339)) + + // zero is the key used in beam.AddFixedKey which will be applied on the main input. + id, externalData := 0, "some fake data that changed at "+time.Now().Format(time.RFC3339) + + emit(id, externalData) +} + +// process simulates processing of main input. It reads side input by key +func process(ctx context.Context, k int, v []byte, side func(int) func(*string) bool) { + log.Infof(ctx, "Processing (key:%d,value:%q)", k, v) + + iter := side(k) + + var externalData []string + var externalDatum string + for iter(&externalDatum) { + externalData = append(externalData, externalDatum) + } + + log.Infof(ctx, "Processing (key:%d,value:%q) with external data %q", k, v, strings.Join(externalData, ",")) +} + +func fatalf(err error, format string, args ...interface{}) { + if err != nil { + log.Fatalf(context.TODO(), format, args...) + } +} + +func main() { + var inputTopic, periodicSequenceStart, periodicSequenceEnd string + var periodicSequenceInterval time.Duration + + now := time.Now() + + flag.StringVar(&periodicSequenceStart, "periodic_sequence_start", now.Add(-1*time.Hour).Format(time.RFC3339), + "The time at which to start the periodic sequence.") + + flag.StringVar(&periodicSequenceEnd, "periodic_sequence_end", now.Add(100*time.Hour).Format(time.RFC3339), + "The time at which to end the periodic sequence.") + + flag.DurationVar(&periodicSequenceInterval, "periodic_sequence_interval", 1*time.Minute, + "The interval between periodic sequence output.") + + flag.StringVar(&inputTopic, "input_topic", "input", + "The PubSub topic from which to read the main input data.") + + flag.Parse() + beam.Init() + ctx := context.Background() + p, s := beam.NewPipelineWithRoot() + + project := gcpopts.GetProject(ctx) + client, err := pubsub.NewClient(ctx, project) + fatalf(err, "Failed to create client: %v", err) + _, err = pubsubx.EnsureTopic(ctx, client, inputTopic) + fatalf(err, "Failed to ensure topic: %v", err) + + mainInput := beam.WindowInto( + s, + window.NewFixedWindows(periodicSequenceInterval), + beam.AddFixedKey( // simulate keyed data by adding a fixed key + s, + pubsubio.Read( + s, + project, + inputTopic, + nil, + ), + ), + beam.Trigger(trigger.Repeat(trigger.Always())), + beam.PanesDiscard(), + ) + + startTime, _ := time.Parse(time.RFC3339, periodicSequenceStart) + endTime, _ := time.Parse(time.RFC3339, periodicSequenceEnd) + sideInput := beam.WindowInto(s, window.NewFixedWindows(periodicSequenceInterval), + beam.ParDo( + s, + update, + periodic.Impulse( + s, + startTime, + endTime, + periodicSequenceInterval, + ), + ), + beam.Trigger(trigger.Repeat(trigger.Always())), + beam.PanesDiscard(), + ) Review Comment: I thought of making the side input window larger. Do you think it is worth making that change, taking another configuration to specify the side input window size? I'm also curious to know what you mean by > this behavior isn't yet enabled by default in the Go SDK ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) + return offsetrange.Restriction{ + Start: int64(0), + End: int64(totalOutputs), + } +} + +func (fn *sequenceGenDoFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker { + return sdf.NewLockRTracker(offsetrange.NewTracker(rest)) +} + +func (fn *sequenceGenDoFn) RestrictionSize(_ SequenceDefinition, rest offsetrange.Restriction) float64 { + return rest.Size() +} + +func (fn *sequenceGenDoFn) SplitRestriction(_ SequenceDefinition, rest offsetrange.Restriction) []offsetrange.Restriction { + return []offsetrange.Restriction{rest} +} + +// TruncateRestriction immediately truncates the entire restrication. +func (fn *sequenceGenDoFn) TruncateRestriction(_ *sdf.LockRTracker, _ SequenceDefinition) offsetrange.Restriction { + return offsetrange.Restriction{} +} + +func (fn *sequenceGenDoFn) CreateWatermarkEstimator() *sdf.ManualWatermarkEstimator { + return &sdf.ManualWatermarkEstimator{} +} + +func (fn *sequenceGenDoFn) ProcessElement(ctx context.Context, we *sdf.ManualWatermarkEstimator, rt *sdf.LockRTracker, sd SequenceDefinition, emit func(beam.EventTime, int64)) (sdf.ProcessContinuation, error) { + currentOutputIndex := rt.GetRestriction().(offsetrange.Restriction).Start + currentOutputTimestamp := sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime := fn.now() + we.UpdateWatermark(currentOutputTimestamp) + for currentOutputTimestamp.Before(currentTime) { + if rt.TryClaim(currentOutputIndex) { + emit(mtime.FromTime(currentOutputTimestamp), currentOutputTimestamp.UnixMilli()) + currentOutputIndex += 1 + currentOutputTimestamp = sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime = fn.now() + we.UpdateWatermark(currentOutputTimestamp) + } else if err := rt.GetError(); err != nil || rt.IsDone() { + // Stop processing on error or completion + return sdf.StopProcessing(), rt.GetError() + } else { + return sdf.ResumeProcessingIn(sd.Interval), nil + } + } + + return sdf.ResumeProcessingIn(time.Until(currentOutputTimestamp)), nil +} + +type impulseConfig struct { + ApplyWindow bool + + now func() time.Time +} + +type impulseOption func(*impulseConfig) error + +// ImpulseOption is a function that configures an [Impulse] transform. +type ImpulseOption = impulseOption + +// WithApplyWindow configures the [Impulse] transform to apply a fixed window +// transform to the output PCollection. +func WithApplyWindow() ImpulseOption { Review Comment: Sounds good to me! I've gone ahead and removed the functional options entirely in favour of a simple `applyWindow bool`. This also makes some of the other comments moot. ########## sdks/go/examples/slowly_updating_side_input/slowly_updating_side_input.go: ########## @@ -0,0 +1,145 @@ +// 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 main + +import ( + "context" + "flag" + "strings" + "time" + + "cloud.google.com/go/pubsub" + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/pubsubio" + "github.com/apache/beam/sdks/v2/go/pkg/beam/log" + "github.com/apache/beam/sdks/v2/go/pkg/beam/options/gcpopts" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" + _ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/dataflow" + "github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/periodic" + "github.com/apache/beam/sdks/v2/go/pkg/beam/util/pubsubx" +) + +func init() { + register.Function4x0(update) + register.Function4x0(process) + register.Emitter2[int, string]() + register.Iter1[string]() +} + +// update simulates an external call to get data for the side input. +func update(ctx context.Context, t beam.EventTime, i int64, emit func(int, string)) { + log.Infof(ctx, "Making external call %d at %s", i, t.ToTime().Format(time.RFC3339)) + + // zero is the key used in beam.AddFixedKey which will be applied on the main input. + id, externalData := 0, "some fake data that changed at "+time.Now().Format(time.RFC3339) + + emit(id, externalData) +} + +// process simulates processing of main input. It reads side input by key +func process(ctx context.Context, k int, v []byte, side func(int) func(*string) bool) { + log.Infof(ctx, "Processing (key:%d,value:%q)", k, v) + + iter := side(k) + + var externalData []string + var externalDatum string + for iter(&externalDatum) { + externalData = append(externalData, externalDatum) + } + + log.Infof(ctx, "Processing (key:%d,value:%q) with external data %q", k, v, strings.Join(externalData, ",")) +} + +func fatalf(err error, format string, args ...interface{}) { + if err != nil { + log.Fatalf(context.TODO(), format, args...) + } +} + +func main() { + var inputTopic, periodicSequenceStart, periodicSequenceEnd string + var periodicSequenceInterval time.Duration + + now := time.Now() + + flag.StringVar(&periodicSequenceStart, "periodic_sequence_start", now.Add(-1*time.Hour).Format(time.RFC3339), + "The time at which to start the periodic sequence.") + + flag.StringVar(&periodicSequenceEnd, "periodic_sequence_end", now.Add(100*time.Hour).Format(time.RFC3339), + "The time at which to end the periodic sequence.") + + flag.DurationVar(&periodicSequenceInterval, "periodic_sequence_interval", 1*time.Minute, + "The interval between periodic sequence output.") + + flag.StringVar(&inputTopic, "input_topic", "input", + "The PubSub topic from which to read the main input data.") + + flag.Parse() + beam.Init() + ctx := context.Background() + p, s := beam.NewPipelineWithRoot() + + project := gcpopts.GetProject(ctx) + client, err := pubsub.NewClient(ctx, project) + fatalf(err, "Failed to create client: %v", err) + _, err = pubsubx.EnsureTopic(ctx, client, inputTopic) + fatalf(err, "Failed to ensure topic: %v", err) + + mainInput := beam.WindowInto( + s, + window.NewFixedWindows(periodicSequenceInterval), + beam.AddFixedKey( // simulate keyed data by adding a fixed key + s, + pubsubio.Read( + s, + project, + inputTopic, + nil, + ), + ), + beam.Trigger(trigger.Repeat(trigger.Always())), + beam.PanesDiscard(), + ) Review Comment: Gotcha! I do agree and I'll keep it in mind for next time. ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) + return offsetrange.Restriction{ + Start: int64(0), + End: int64(totalOutputs), + } +} + +func (fn *sequenceGenDoFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker { + return sdf.NewLockRTracker(offsetrange.NewTracker(rest)) +} + +func (fn *sequenceGenDoFn) RestrictionSize(_ SequenceDefinition, rest offsetrange.Restriction) float64 { + return rest.Size() +} + +func (fn *sequenceGenDoFn) SplitRestriction(_ SequenceDefinition, rest offsetrange.Restriction) []offsetrange.Restriction { + return []offsetrange.Restriction{rest} +} + +// TruncateRestriction immediately truncates the entire restrication. +func (fn *sequenceGenDoFn) TruncateRestriction(_ *sdf.LockRTracker, _ SequenceDefinition) offsetrange.Restriction { + return offsetrange.Restriction{} +} + +func (fn *sequenceGenDoFn) CreateWatermarkEstimator() *sdf.ManualWatermarkEstimator { + return &sdf.ManualWatermarkEstimator{} +} + +func (fn *sequenceGenDoFn) ProcessElement(ctx context.Context, we *sdf.ManualWatermarkEstimator, rt *sdf.LockRTracker, sd SequenceDefinition, emit func(beam.EventTime, int64)) (sdf.ProcessContinuation, error) { + currentOutputIndex := rt.GetRestriction().(offsetrange.Restriction).Start + currentOutputTimestamp := sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime := fn.now() + we.UpdateWatermark(currentOutputTimestamp) + for currentOutputTimestamp.Before(currentTime) { + if rt.TryClaim(currentOutputIndex) { + emit(mtime.FromTime(currentOutputTimestamp), currentOutputTimestamp.UnixMilli()) + currentOutputIndex += 1 + currentOutputTimestamp = sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime = fn.now() + we.UpdateWatermark(currentOutputTimestamp) + } else if err := rt.GetError(); err != nil || rt.IsDone() { + // Stop processing on error or completion + return sdf.StopProcessing(), rt.GetError() + } else { + return sdf.ResumeProcessingIn(sd.Interval), nil + } + } + + return sdf.ResumeProcessingIn(time.Until(currentOutputTimestamp)), nil +} + +type impulseConfig struct { + ApplyWindow bool + + now func() time.Time +} + +type impulseOption func(*impulseConfig) error + +// ImpulseOption is a function that configures an [Impulse] transform. +type ImpulseOption = impulseOption + +// WithApplyWindow configures the [Impulse] transform to apply a fixed window +// transform to the output PCollection. +func WithApplyWindow() ImpulseOption { + return func(o *impulseConfig) error { + o.ApplyWindow = true + return nil + } +} + +func withNowFunc(now func() time.Time) ImpulseOption { Review Comment: Removed. ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## Review Comment: That's great! I couldn't get it working on the direct runner, but didn't try using the new prism explicitly. Have now added two tests `TestImpulse` and `TestSequence` and it looks like that's working. Let me know what you think. ########## sdks/go/pkg/beam/transforms/periodic/periodic.go: ########## @@ -0,0 +1,211 @@ +// 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 periodic contains transformations for generating periodic sequences. +package periodic + +import ( + "context" + "fmt" + "math" + "reflect" + "time" + + "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/mtime" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window" + "github.com/apache/beam/sdks/v2/go/pkg/beam/core/sdf" + "github.com/apache/beam/sdks/v2/go/pkg/beam/io/rtrackers/offsetrange" + "github.com/apache/beam/sdks/v2/go/pkg/beam/register" +) + +func init() { + register.DoFn5x2[context.Context, *sdf.ManualWatermarkEstimator, *sdf.LockRTracker, SequenceDefinition, + func(beam.EventTime, int64), + sdf.ProcessContinuation, error](&sequenceGenDoFn{}) + register.Emitter2[beam.EventTime, int64]() + beam.RegisterType(reflect.TypeOf(SequenceDefinition{})) +} + +// SequenceDefinition holds the configuration for generating a sequence of +// timestamped elements at an interval. +type SequenceDefinition struct { + Interval time.Duration + Start time.Time + End time.Time +} + +type sequenceGenDoFn struct { + now func() time.Time +} + +func (fn *sequenceGenDoFn) Setup() { + if fn.now == nil { + fn.now = time.Now + } +} + +func (fn *sequenceGenDoFn) CreateInitialRestriction(sd SequenceDefinition) offsetrange.Restriction { + totalOutputs := math.Ceil(float64(sd.End.Sub(sd.Start) / sd.Interval)) + return offsetrange.Restriction{ + Start: int64(0), + End: int64(totalOutputs), + } +} + +func (fn *sequenceGenDoFn) CreateTracker(rest offsetrange.Restriction) *sdf.LockRTracker { + return sdf.NewLockRTracker(offsetrange.NewTracker(rest)) +} + +func (fn *sequenceGenDoFn) RestrictionSize(_ SequenceDefinition, rest offsetrange.Restriction) float64 { + return rest.Size() +} + +func (fn *sequenceGenDoFn) SplitRestriction(_ SequenceDefinition, rest offsetrange.Restriction) []offsetrange.Restriction { + return []offsetrange.Restriction{rest} +} + +// TruncateRestriction immediately truncates the entire restrication. +func (fn *sequenceGenDoFn) TruncateRestriction(_ *sdf.LockRTracker, _ SequenceDefinition) offsetrange.Restriction { + return offsetrange.Restriction{} +} + +func (fn *sequenceGenDoFn) CreateWatermarkEstimator() *sdf.ManualWatermarkEstimator { + return &sdf.ManualWatermarkEstimator{} +} + +func (fn *sequenceGenDoFn) ProcessElement(ctx context.Context, we *sdf.ManualWatermarkEstimator, rt *sdf.LockRTracker, sd SequenceDefinition, emit func(beam.EventTime, int64)) (sdf.ProcessContinuation, error) { + currentOutputIndex := rt.GetRestriction().(offsetrange.Restriction).Start + currentOutputTimestamp := sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime := fn.now() + we.UpdateWatermark(currentOutputTimestamp) + for currentOutputTimestamp.Before(currentTime) { + if rt.TryClaim(currentOutputIndex) { + emit(mtime.FromTime(currentOutputTimestamp), currentOutputTimestamp.UnixMilli()) + currentOutputIndex += 1 + currentOutputTimestamp = sd.Start.Add(sd.Interval * time.Duration(currentOutputIndex)) + currentTime = fn.now() + we.UpdateWatermark(currentOutputTimestamp) + } else if err := rt.GetError(); err != nil || rt.IsDone() { + // Stop processing on error or completion + return sdf.StopProcessing(), rt.GetError() + } else { + return sdf.ResumeProcessingIn(sd.Interval), nil + } + } + + return sdf.ResumeProcessingIn(time.Until(currentOutputTimestamp)), nil +} + +type impulseConfig struct { + ApplyWindow bool + + now func() time.Time +} + +type impulseOption func(*impulseConfig) error + +// ImpulseOption is a function that configures an [Impulse] transform. +type ImpulseOption = impulseOption Review Comment: Since we've opted for using `applyWindow bool` this has now been removed. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
