You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "slilichenko (via GitHub)" <gi...@apache.org> on 2024/09/23 22:09:54 UTC
[PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
slilichenko opened a new pull request, #32540:
URL: https://github.com/apache/beam/pull/32540
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] 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://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792370457
##########
CHANGES.md:
##########
@@ -68,6 +68,7 @@
* Dataflow worker can install packages from Google Artifact Registry Python repositories (Python) ([#32123](https://github.com/apache/beam/issues/32123)).
* Added support for Zstd codec in SerializableAvroCodecFactory (Java) ([#32349](https://github.com/apache/beam/issues/32349))
+* Added support for processing events which use a global sequence to "ordered" extension (Java)
Review Comment:
Removed this line from 2.60.0 section. Will add to 2.61.0 in the next PR.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792497643
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.extensions.ordered.ProcessingState.ProcessingStateCoder;
+import org.apache.beam.sdk.state.OrderedListState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Main Stateful DoFn used to process events in the global sequence mode.
+ *
+ * @param <EventT>
+ * @param <EventKeyT>
+ * @param <ResultT>
+ * @param <StateT>
+ */
+class GlobalSequencesProcessorDoFn<
+ EventT, EventKeyT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends ProcessorDoFn<EventT, EventKeyT, ResultT, StateT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlobalSequencesProcessorDoFn.class);
+
+ private static final String BATCH_EMISSION_TIMER = "batchTimer";
+
+ @TimerId(BATCH_EMISSION_TIMER)
+ @SuppressWarnings("unused")
+ private final TimerSpec batchTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
Review Comment:
Yeah. I guess we could also look at the input allowed lateness and use that to key the timer though.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792387582
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
+ private @Nullable Long initialSequence = null;
+
+ public void add(long sequence, Instant timestamp, boolean isInitialSequence) {
+ if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {
+ throw new IllegalStateException(
+ "There are different initial sequences detected: "
+ + initialSequence
+ + " and "
+ + sequence);
+ }
+
+ if (isInitialSequence) {
+ this.initialSequence = sequence;
+ clearRangesBelowInitialSequence(sequence, timestamp);
+ } else if (initialSequence != null && sequence <= initialSequence) {
+ // No need to add anything lower than the initial sequence to the accumulator.
+ return;
+ }
+
+ long lowerBound = sequence, upperBound = sequence;
Review Comment:
Agree on the same outcome. I am afraid that computationally it's going to be more expensive. Requires more lookups, creating a stream and iterator, etc. For high volume processing it can add 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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1791932081
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
Review Comment:
I think this should be pulled from the input windowing strategy, right? We can pull it from https://github.com/slilichenko/beam/blob/3035e5a8fa05ca2c22eb7097226a9312f3bd6481/sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowingStrategy.java#L137
Otherwise, allowed late data might be dropped
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/UnprocessedEvent.java:
##########
@@ -68,6 +68,7 @@ static <EventT> UnprocessedEvent<EventT> create(
return new AutoValue_UnprocessedEvent<>(event, reason, failureDetails);
}
+ // TODO: do we need another reason, "before the initial event"?
Review Comment:
Probably this would be good to add
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
+ .triggering(
+ Repeatedly.forever(
+ AfterFirst.of(
+ AfterPane.elementCountAtLeast(1),
Review Comment:
If so, is there a point in emitting after processing time?
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
+ .triggering(
+ Repeatedly.forever(
+ AfterFirst.of(
+ AfterPane.elementCountAtLeast(1),
Review Comment:
Will this just cause this trigger to fire repeatedly after every element it sees?
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.extensions.ordered.ProcessingState.ProcessingStateCoder;
+import org.apache.beam.sdk.state.OrderedListState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Main Stateful DoFn used to process events in the global sequence mode.
+ *
+ * @param <EventT>
+ * @param <EventKeyT>
+ * @param <ResultT>
+ * @param <StateT>
+ */
+class GlobalSequencesProcessorDoFn<
+ EventT, EventKeyT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends ProcessorDoFn<EventT, EventKeyT, ResultT, StateT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlobalSequencesProcessorDoFn.class);
+
+ private static final String BATCH_EMISSION_TIMER = "batchTimer";
+
+ @TimerId(BATCH_EMISSION_TIMER)
+ @SuppressWarnings("unused")
+ private final TimerSpec batchTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
Review Comment:
Hm, noting that I think the usage of this timer will cause issues if the input has allowed lateness. We may need to disallow handling late data here
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792530303
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
+ private @Nullable Long initialSequence = null;
+
+ public void add(long sequence, Instant timestamp, boolean isInitialSequence) {
+ if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {
+ throw new IllegalStateException(
+ "There are different initial sequences detected: "
+ + initialSequence
+ + " and "
+ + sequence);
+ }
+
+ if (isInitialSequence) {
+ this.initialSequence = sequence;
+ clearRangesBelowInitialSequence(sequence, timestamp);
Review Comment:
The data is not discarded in this case; it's output to the DLQ output with "duplicate" reason - ProcessorDoFn, line 349 (for some reason can't create a permalink in PR).
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792492426
##########
CHANGES.md:
##########
@@ -68,6 +68,7 @@
* Dataflow worker can install packages from Google Artifact Registry Python repositories (Python) ([#32123](https://github.com/apache/beam/issues/32123)).
* Added support for Zstd codec in SerializableAvroCodecFactory (Java) ([#32349](https://github.com/apache/beam/issues/32349))
+* Added support for processing events which use a global sequence to "ordered" extension (Java)
Review Comment:
If you pull in the latest changes from master we can still add this to changes.md in this pr. It is ok to defer if you'd prefer though
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1793570230
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
Review Comment:
LGTM, thank you
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792624984
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.extensions.ordered.ProcessingState.ProcessingStateCoder;
+import org.apache.beam.sdk.state.OrderedListState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Main Stateful DoFn used to process events in the global sequence mode.
+ *
+ * @param <EventT>
+ * @param <EventKeyT>
+ * @param <ResultT>
+ * @param <StateT>
+ */
+class GlobalSequencesProcessorDoFn<
+ EventT, EventKeyT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends ProcessorDoFn<EventT, EventKeyT, ResultT, StateT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlobalSequencesProcessorDoFn.class);
+
+ private static final String BATCH_EMISSION_TIMER = "batchTimer";
+
+ @TimerId(BATCH_EMISSION_TIMER)
+ @SuppressWarnings("unused")
+ private final TimerSpec batchTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
Review Comment:
Good point! Added the logic to capture the lateness of the input and add it to the max timestamp.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792628217
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/UnprocessedEvent.java:
##########
@@ -68,6 +68,7 @@ static <EventT> UnprocessedEvent<EventT> create(
return new AutoValue_UnprocessedEvent<>(event, reason, failureDetails);
}
+ // TODO: do we need another reason, "before the initial event"?
Review Comment:
Done
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1793569402
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
+ private @Nullable Long initialSequence = null;
+
+ public void add(long sequence, Instant timestamp, boolean isInitialSequence) {
+ if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {
+ throw new IllegalStateException(
+ "There are different initial sequences detected: "
+ + initialSequence
+ + " and "
+ + sequence);
+ }
+
+ if (isInitialSequence) {
+ this.initialSequence = sequence;
+ clearRangesBelowInitialSequence(sequence, timestamp);
Review Comment:
Oh I see, you're right. 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792365236
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ContiguousSequenceRange.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** A range of contiguous event sequences and the latest timestamp of the events in the range. */
+@AutoValue
+public abstract class ContiguousSequenceRange {
+ public static final ContiguousSequenceRange EMPTY =
+ ContiguousSequenceRange.of(
+ Long.MIN_VALUE, Long.MIN_VALUE, Instant.ofEpochMilli(Long.MIN_VALUE));
Review Comment:
Thanks - makes sense. I made ContiguousSequenceRange open ended.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792395233
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
+ .triggering(
+ Repeatedly.forever(
+ AfterFirst.of(
+ AfterPane.elementCountAtLeast(1),
Review Comment:
What you are saying - it will trigger on every element, regardless of the other time based trigger? That's not the intention, I probably need to specify something larger than 1 to avoid constant recalculation of the side input.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm merged PR #32540:
URL: https://github.com/apache/beam/pull/32540
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792567838
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
Review Comment:
Fixed - thank you!
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792369225
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/ContiguousSequenceRange.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.InstantCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** A range of contiguous event sequences and the latest timestamp of the events in the range. */
+@AutoValue
+public abstract class ContiguousSequenceRange {
+ public static final ContiguousSequenceRange EMPTY =
+ ContiguousSequenceRange.of(
+ Long.MIN_VALUE, Long.MIN_VALUE, Instant.ofEpochMilli(Long.MIN_VALUE));
+
+ /** @return inclusive starting sequence */
+ public abstract long getStart();
Review Comment:
It's not only for the combiner; it's used in the processing DoFn to DLQ the elements below the range.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792496194
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
+ private @Nullable Long initialSequence = null;
+
+ public void add(long sequence, Instant timestamp, boolean isInitialSequence) {
+ if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {
+ throw new IllegalStateException(
+ "There are different initial sequences detected: "
+ + initialSequence
+ + " and "
+ + sequence);
+ }
+
+ if (isInitialSequence) {
+ this.initialSequence = sequence;
+ clearRangesBelowInitialSequence(sequence, timestamp);
+ } else if (initialSequence != null && sequence <= initialSequence) {
+ // No need to add anything lower than the initial sequence to the accumulator.
+ return;
+ }
+
+ long lowerBound = sequence, upperBound = sequence;
Review Comment:
Fair enough, SGTM
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792396463
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequencesProcessorDoFn.java:
##########
@@ -0,0 +1,272 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.extensions.ordered.ProcessingState.ProcessingStateCoder;
+import org.apache.beam.sdk.state.OrderedListState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Main Stateful DoFn used to process events in the global sequence mode.
+ *
+ * @param <EventT>
+ * @param <EventKeyT>
+ * @param <ResultT>
+ * @param <StateT>
+ */
+class GlobalSequencesProcessorDoFn<
+ EventT, EventKeyT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends ProcessorDoFn<EventT, EventKeyT, ResultT, StateT> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GlobalSequencesProcessorDoFn.class);
+
+ private static final String BATCH_EMISSION_TIMER = "batchTimer";
+
+ @TimerId(BATCH_EMISSION_TIMER)
+ @SuppressWarnings("unused")
+ private final TimerSpec batchTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
Review Comment:
Because this type of timer doesn't honor the lateness?
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "damccorm (via GitHub)" <gi...@apache.org>.
damccorm commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792495329
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
+ private @Nullable Long initialSequence = null;
+
+ public void add(long sequence, Instant timestamp, boolean isInitialSequence) {
+ if (isInitialSequence && this.initialSequence != null && sequence != this.initialSequence) {
+ throw new IllegalStateException(
+ "There are different initial sequences detected: "
+ + initialSequence
+ + " and "
+ + sequence);
+ }
+
+ if (isInitialSequence) {
+ this.initialSequence = sequence;
+ clearRangesBelowInitialSequence(sequence, timestamp);
Review Comment:
Ack - maybe we should at least warn? In this case we are intentionally discarding data, which might be unexpected
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792540453
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/combiner/SequenceRangeAccumulator.java:
##########
@@ -0,0 +1,278 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered.combiner;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange;
+import org.apache.commons.lang3.tuple.Pair;
+import org.checkerframework.checker.initialization.qual.Initialized;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.UnknownKeyFor;
+import org.joda.time.Instant;
+
+/** Default accumulator used to combine sequence ranges. */
+public class SequenceRangeAccumulator {
+
+ private static Instant max(Instant a, Instant b) {
+ return a.isAfter(b) ? a : b;
+ }
+
+ private final TreeMap<Long, Pair<Long, Instant>> data = new TreeMap<>();
Review Comment:
Added the comments. I kept the variable names the same; let me know please if the comments are not sufficent.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792619042
##########
sdks/java/extensions/ordered/src/main/java/org/apache/beam/sdk/extensions/ordered/GlobalSequenceTracker.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 org.apache.beam.sdk.extensions.ordered;
+
+import org.apache.beam.sdk.extensions.ordered.ContiguousSequenceRange.CompletedSequenceRangeCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.AfterFirst;
+import org.apache.beam.sdk.transforms.windowing.AfterPane;
+import org.apache.beam.sdk.transforms.windowing.AfterProcessingTime;
+import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.WindowFn;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+
+/**
+ * PTransform to produce the side input of the maximum contiguous range of sequence numbers.
+ *
+ * @param <EventKeyT> type of event key
+ * @param <EventT> type of event
+ * @param <ResultT> type of processing result
+ * @param <StateT> type of state
+ */
+class GlobalSequenceTracker<
+ EventKeyT, EventT, ResultT, StateT extends MutableState<EventT, ResultT>>
+ extends PTransform<
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>,
+ PCollectionView<ContiguousSequenceRange>> {
+
+ private final Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer;
+ private final @Nullable Duration frequencyOfGeneration;
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = null;
+ }
+
+ public GlobalSequenceTracker(
+ Combine.GloballyAsSingletonView<
+ TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ContiguousSequenceRange>
+ sideInputProducer,
+ Duration globalSequenceGenerationFrequency) {
+ this.sideInputProducer = sideInputProducer;
+ this.frequencyOfGeneration = globalSequenceGenerationFrequency;
+ }
+
+ @Override
+ public PCollectionView<ContiguousSequenceRange> expand(
+ PCollection<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>> input) {
+ input
+ .getPipeline()
+ .getCoderRegistry()
+ .registerCoderForClass(ContiguousSequenceRange.class, CompletedSequenceRangeCoder.of());
+
+ if (frequencyOfGeneration != null) {
+ // This branch will only be executed in case of streaming pipelines.
+ // For batch pipelines the side input should only be computed once.
+ input =
+ input.apply(
+ "Triggering Setup",
+ // Reproduce the windowing of the input PCollection, but change the triggering
+ // in order to create a slowing changing side input
+ Window.<TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>>into(
+ (WindowFn<? super TimestampedValue<KV<EventKeyT, KV<Long, EventT>>>, ?>)
+ input.getWindowingStrategy().getWindowFn())
+ .accumulatingFiredPanes()
+ // TODO: verify that we don't need to have the lateness parameterized
+ .withAllowedLateness(Duration.ZERO)
+ .triggering(
+ Repeatedly.forever(
+ AfterFirst.of(
+ AfterPane.elementCountAtLeast(1),
Review Comment:
I fixed this class - made the number of elements a parameter to the handler (with default of 1000). Unit tests run green.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #32540:
URL: https://github.com/apache/beam/pull/32540#discussion_r1792522739
##########
CHANGES.md:
##########
@@ -68,6 +68,7 @@
* Dataflow worker can install packages from Google Artifact Registry Python repositories (Python) ([#32123](https://github.com/apache/beam/issues/32123)).
* Added support for Zstd codec in SerializableAvroCodecFactory (Java) ([#32349](https://github.com/apache/beam/issues/32349))
+* Added support for processing events which use a global sequence to "ordered" extension (Java)
Review Comment:
OK, added a note to 2.61.0 section with a link to this PR.
--
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: [PR] Add support for global sequence processing to the "ordered" extension in Java SDK [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32540:
URL: https://github.com/apache/beam/pull/32540#issuecomment-2369844163
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @m-trieu for label java.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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