You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "celeste-zeng (via GitHub)" <gi...@apache.org> on 2024/06/04 07:38:56 UTC
[PR] Add a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
celeste-zeng opened a new pull request, #31490:
URL: https://github.com/apache/beam/pull/31490
The DataflowGroupByKey is identical to the Beam GroupByKey transform except for having a allowDuplicates property plumbed from the Redistribute transform.
------------------------
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1631669585
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
+ * representations.
+ */
+@SuppressWarnings({
+ "nullness", // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Tried to remove it then multiple checks start to fail.
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Then reified.apply(
originalTransform.getAllowDuplicates()
? DataflowGroupByKey.createWithAllowDuplicates()
: DataflowGroupByKey.create());
will cause check failure due to imcompatible types for .apply method.
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1634180092
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Moved the suppress to the corresponding place because I still can't find a way to resolve the incompatible types:
found : @Initialized @NonNull PCollection<@Initialized @NonNull KV<K extends @Initialized @Nullable Object, @Initialized @NonNull Iterable<@Initialized @Nullable ValueInSingleWindow<V extends @Initialized @Nullable Object>>>>
required: @UnknownInitialization @Nullable PCollection<@Initialized @NonNull KV<K extends @Initialized @Nullable Object, @Initialized @NonNull Iterable<@Initialized @NonNull ValueInSingleWindow<V extends @Initialized @Nullable Object>>>>
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1631555066
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
don't suppress in new files
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1633336029
##########
model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto:
##########
@@ -327,6 +327,11 @@ message StandardPTransforms {
//
// Payload: none
TO_STRING = 8 [(beam_urn) = "beam:transform:to_string:v1"];
+
+ // Specialized implementation of GroupByKey for translating Redistribute transform into
+ // Dataflow service protos.
+ DATAFLOW_GROUP_BY_KEY = 9
Review Comment:
I don't think you need this, because you should not have this transform in a portable graph, only in the v1beta3 (which is a pure Java override straight to v1beta3 and exists without the proto)
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DataflowGroupByKey.java:
##########
@@ -0,0 +1,157 @@
+/*
+ * 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.transforms;
+
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.Coder.NonDeterministicException;
+import org.apache.beam.sdk.coders.IterableCoder;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollection.IsBounded;
+import org.apache.beam.sdk.values.WindowingStrategy;
+
+/**
+ * Specialized implementation of {@code GroupByKey} for translating Redistribute transform into
+ * Dataflow service protos.
+ */
+public class DataflowGroupByKey<K, V>
+ extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Iterable<V>>>> {
+
+ // Plumbed from Redistribute transform.
+ private final boolean allowDuplicates;
+
+ private DataflowGroupByKey(boolean allowDuplicates) {
+ this.allowDuplicates = allowDuplicates;
+ }
+
+ /**
+ * Returns a {@code DataflowGroupByKey<K, V>} {@code PTransform}.
+ *
+ * @param <K> the type of the keys of the input and output {@code PCollection}s
+ * @param <V> the type of the values of the input {@code PCollection} and the elements of the
+ * {@code Iterable}s in the output {@code PCollection}
+ */
+ public static <K, V> DataflowGroupByKey<K, V> create() {
+ return new DataflowGroupByKey<>(false);
+ }
+
+ /**
+ * Returns a {@code DataflowGroupByKey<K, V>} {@code PTransform} that its output can have
+ * duplicated elements.
+ *
+ * @param <K> the type of the keys of the input and output {@code PCollection}s
+ * @param <V> the type of the values of the input {@code PCollection} and the elements of the
+ * {@code Iterable}s in the output {@code PCollection}
+ */
+ public static <K, V> DataflowGroupByKey<K, V> createWithAllowDuplicates() {
+ return new DataflowGroupByKey<>(true);
+ }
+
+ /** Returns whether it allows duplicated elements in the output. */
+ public boolean allowDuplicates() {
+ return allowDuplicates;
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+
+ public static void applicableTo(PCollection<?> input) {
+ WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
+ // Verify that the input PCollection is bounded, or that there is windowing/triggering being
+ // used. Without this, the watermark (at end of global window) will never be reached.
+ if (windowingStrategy.getWindowFn() instanceof GlobalWindows
+ && windowingStrategy.getTrigger() instanceof DefaultTrigger
+ && input.isBounded() != IsBounded.BOUNDED) {
+ throw new IllegalStateException(
+ "DataflowGroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow"
+ + " without a trigger. Use a Window.into or Window.triggering transform prior to"
+ + " DataflowGroupByKey.");
+ }
+ }
+
+ public WindowingStrategy<?, ?> updateWindowingStrategy(WindowingStrategy<?, ?> inputStrategy) {
+ // If the WindowFn was merging, set the bit to indicate it is already merged.
+ // Switch to the continuation trigger associated with the current trigger.
+ return inputStrategy
+ .withAlreadyMerged(!inputStrategy.getWindowFn().isNonMerging())
+ .withTrigger(inputStrategy.getTrigger().getContinuationTrigger());
+ }
+
+ @Override
+ public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) {
+ applicableTo(input);
+
+ // Verify that the input Coder<KV<K, V>> is a KvCoder<K, V>, and that
+ // the key coder is deterministic.
+ Coder<K> keyCoder = getKeyCoder(input.getCoder());
+ try {
+ keyCoder.verifyDeterministic();
+ } catch (NonDeterministicException e) {
+ throw new IllegalStateException(
+ "the keyCoder of a DataflowGroupByKey must be deterministic", e);
+ }
+
+ // This primitive operation groups by the combination of key and window,
+ // merging windows as needed, using the windows assigned to the
+ // key/value input elements and the window merge operation of the
+ // window function associated with the input PCollection.
+ return PCollection.createPrimitiveOutputInternal(
+ input.getPipeline(),
+ updateWindowingStrategy(input.getWindowingStrategy()),
+ input.isBounded(),
+ getOutputKvCoder(input.getCoder()));
+ }
+
+ /**
+ * Returns the {@code Coder} of the input to this transform, which should be a {@code KvCoder}.
+ */
+ @SuppressWarnings("unchecked")
+ static <K, V> KvCoder<K, V> getInputKvCoder(Coder<?> inputCoder) {
+ if (!(inputCoder instanceof KvCoder)) {
+ throw new IllegalStateException("DataflowGroupByKey requires its input to use KvCoder");
+ }
+ return (KvCoder<K, V>) inputCoder;
+ }
+
+ /////////////////////////////////////////////////////////////////////////////
+
+ /**
+ * Returns the {@code Coder} of the keys of the input to this transform, which is also used as the
+ * {@code Coder} of the keys of the output of this transform.
+ */
+ public static <K, V> Coder<K> getKeyCoder(Coder<KV<K, V>> inputCoder) {
Review Comment:
These don't need to be public, right?
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
What are the incompatible types? Basically, treat nullness the same way you would treat any other type error. You can suppress at a very fine local level if absolutely necessary. But in this case I think you just need to add a null check if there is a nullable value probably.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/PTransformTranslation.java:
##########
@@ -177,6 +179,9 @@ public class PTransformTranslation {
checkState(FLATTEN_TRANSFORM_URN.equals(getUrn(StandardPTransforms.Primitives.FLATTEN)));
checkState(
GROUP_BY_KEY_TRANSFORM_URN.equals(getUrn(StandardPTransforms.Primitives.GROUP_BY_KEY)));
+ checkState(
Review Comment:
I don't think we need this test, since this transform should never show up in a portable graph. Removing the URN from the proto is a good way to protect us from mistakes, I think.
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
+ * representations.
+ */
+@SuppressWarnings({
+ "nullness", // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Going to have to get them right. You can add null checks where you need them. This isn't an optional warning; it is a type system that protects from errors, including helping to enforce good code approaches. (most coding that has a problem with null checks can be replaced by approaches that have fewer problems)
Treat it the same as any other type error.
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31490:
URL: https://github.com/apache/beam/pull/31490#issuecomment-2146935243
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #31490:
URL: https://github.com/apache/beam/pull/31490#issuecomment-2155294924
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn for label java.
R: @Abacn for label build.
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
Re: [PR] Add a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #31490:
URL: https://github.com/apache/beam/pull/31490#issuecomment-2158977656
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/31490?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 71.17%. Comparing base [(`7a0ad41`)](https://app.codecov.io/gh/apache/beam/commit/7a0ad41935fdf90a2a608fb02c2f9a7eb714db18?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`7082ff2`)](https://app.codecov.io/gh/apache/beam/commit/7082ff2c6df4deb4098c8a992272a408d2142639?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 7 commits behind head on master.
> :exclamation: **Current head 7082ff2 differs from pull request most recent head 6452750**
>
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) reports for the commit 6452750 to get more accurate results.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #31490 +/- ##
============================================
- Coverage 71.18% 71.17% -0.01%
Complexity 3014 3014
============================================
Files 1057 1057
Lines 133906 133906
Branches 3254 3254
============================================
- Hits 95318 95310 -8
- Misses 35450 35458 +8
Partials 3138 3138
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/31490?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1633772722
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
+ * representations.
+ */
+@SuppressWarnings({
+ "nullness", // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Couldn't find a solution to completely remove these suppress, moved them to specific places like what was done in https://github.com/apache/beam/blob/d560aa63d58145546def09aba7db6f0c8f39dbe0/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Redistribute.java#L255 and https://github.com/apache/beam/blob/d560aa63d58145546def09aba7db6f0c8f39dbe0/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Redistribute.java#L275
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1636565338
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
Review Comment:
Is this necessary? Maybe you need it in order to get the URN in some context? Because otherwise obvs it cannot really go to/from runner API since the translator does not preserve all the information. To be clear: I feel like we shouldn't have this file, unless we need it as a technicality.
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1639069589
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
Review Comment:
Actually, I am able to delete this file and just add the contents to DataflowGBK.
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on PR #31490:
URL: https://github.com/apache/beam/pull/31490#issuecomment-2146900892
R:@kennknowles
--
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] Plumb Redistribute "allow duplicates" property to Dataflow [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles merged PR #31490:
URL: https://github.com/apache/beam/pull/31490
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1631554894
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
+ * representations.
+ */
+@SuppressWarnings({
+ "nullness", // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
don't suppress in new files
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1635258681
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+class RedistributeByKeyOverrideFactory<K, V>
+ extends SingleInputOutputOverrideFactory<
+ PCollection<KV<K, V>>, PCollection<KV<K, V>>, RedistributeByKey<K, V>> {
+
+ @Override
+ public PTransformReplacement<PCollection<KV<K, V>>, PCollection<KV<K, V>>>
+ getReplacementTransform(
+ AppliedPTransform<PCollection<KV<K, V>>, PCollection<KV<K, V>>, RedistributeByKey<K, V>>
+ transform) {
+ return PTransformOverrideFactory.PTransformReplacement.of(
+ PTransformReplacements.getSingletonMainInput(transform),
+ new DataflowRedistributeByKey<>(transform.getTransform()));
+ }
+
+ /** Specialized implementation of {@link RedistributeByKey} for Dataflow pipelines. */
+ private static class DataflowRedistributeByKey<K, V>
+ extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, V>>> {
+
+ private final RedistributeByKey<K, V> originalTransform;
+
+ private DataflowRedistributeByKey(RedistributeByKey<K, V> originalTransform) {
+ this.originalTransform = originalTransform;
+ }
+
+ @Override
+ public PCollection<KV<K, V>> expand(PCollection<KV<K, V>> input) {
+ WindowingStrategy<?, ?> originalStrategy = input.getWindowingStrategy();
+ Window<KV<K, V>> rewindow =
+ Window.<KV<K, V>>into(
+ new IdentityWindowFn<>(originalStrategy.getWindowFn().windowCoder()))
+ .triggering(new ReshuffleTrigger<>())
+ .discardingFiredPanes()
+ .withTimestampCombiner(TimestampCombiner.EARLIEST)
+ .withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
+
+ PCollection<KV<K, ValueInSingleWindow<V>>> reified =
+ input
+ .apply("SetIdentityWindow", rewindow)
+ .apply("ReifyOriginalMetadata", Reify.windowsInValue());
+
+ @SuppressWarnings("nullness") // Cannot figure out how to make this typecheck
+ PCollection<KV<K, Iterable<ValueInSingleWindow<V>>>> grouped =
+ reified.apply(
+ originalTransform != null && originalTransform.getAllowDuplicates()
Review Comment:
Maybe try assigning this to a local variable with an explicit type and see what you learn
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1637036486
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/DataflowGroupByKeyTranslation.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.util.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.model.pipeline.v1.RunnerApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Utility methods for translating a {@link DataflowGroupByKey} to and from {@link RunnerApi}
Review Comment:
Yes I think it is needed otherwise I got the following error:
```
Exception in thread "main" java.lang.IllegalArgumentException: Transform Redistribute-Redistribute-RedistributeByKey-DataflowGroupByKey is not a composite transform but does not have a specified URN. inputs {
key: "org.apache.beam.sdk.values.PCollection.<init>:397#c574c40af2bc3c2d"
value: "Redistribute/Redistribute.RedistributeByKey/ReifyOriginalMetadata/ParDo(Anonymous)/ParMultiDo(Anonymous).output"
}
outputs {
key: "org.apache.beam.sdk.values.PCollection.<init>:397#b672aefab09fd4e6"
value: "Redistribute/Redistribute.RedistributeByKey/DataflowGroupByKey.out"
}
unique_name: "Redistribute/Redistribute.RedistributeByKey/DataflowGroupByKey"
```
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1635373131
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+class RedistributeByKeyOverrideFactory<K, V>
+ extends SingleInputOutputOverrideFactory<
+ PCollection<KV<K, V>>, PCollection<KV<K, V>>, RedistributeByKey<K, V>> {
+
+ @Override
+ public PTransformReplacement<PCollection<KV<K, V>>, PCollection<KV<K, V>>>
+ getReplacementTransform(
+ AppliedPTransform<PCollection<KV<K, V>>, PCollection<KV<K, V>>, RedistributeByKey<K, V>>
+ transform) {
+ return PTransformOverrideFactory.PTransformReplacement.of(
+ PTransformReplacements.getSingletonMainInput(transform),
+ new DataflowRedistributeByKey<>(transform.getTransform()));
+ }
+
+ /** Specialized implementation of {@link RedistributeByKey} for Dataflow pipelines. */
+ private static class DataflowRedistributeByKey<K, V>
+ extends PTransform<PCollection<KV<K, V>>, PCollection<KV<K, V>>> {
+
+ private final RedistributeByKey<K, V> originalTransform;
+
+ private DataflowRedistributeByKey(RedistributeByKey<K, V> originalTransform) {
+ this.originalTransform = originalTransform;
+ }
+
+ @Override
+ public PCollection<KV<K, V>> expand(PCollection<KV<K, V>> input) {
+ WindowingStrategy<?, ?> originalStrategy = input.getWindowingStrategy();
+ Window<KV<K, V>> rewindow =
+ Window.<KV<K, V>>into(
+ new IdentityWindowFn<>(originalStrategy.getWindowFn().windowCoder()))
+ .triggering(new ReshuffleTrigger<>())
+ .discardingFiredPanes()
+ .withTimestampCombiner(TimestampCombiner.EARLIEST)
+ .withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
+
+ PCollection<KV<K, ValueInSingleWindow<V>>> reified =
+ input
+ .apply("SetIdentityWindow", rewindow)
+ .apply("ReifyOriginalMetadata", Reify.windowsInValue());
+
+ @SuppressWarnings("nullness") // Cannot figure out how to make this typecheck
+ PCollection<KV<K, Iterable<ValueInSingleWindow<V>>>> grouped =
+ reified.apply(
+ originalTransform != null && originalTransform.getAllowDuplicates()
Review Comment:
I think that did it, 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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "celeste-zeng (via GitHub)" <gi...@apache.org>.
celeste-zeng commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1635136663
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
```
found : @Initialized @NonNull PCollection<@Initialized @NonNull KV<K extends @Initialized @Nullable Object, @Initialized @NonNull Iterable<@Initialized @Nullable ValueInSingleWindow<V extends @Initialized @Nullable Object>>>>
742
required: @UnknownInitialization @Nullable PCollection<@Initialized @NonNull KV<K extends @Initialized @Nullable Object, @Initialized @NonNull Iterable<@Initialized @NonNull ValueInSingleWindow<V extends @Initialized @Nullable Object>>>>
743
/runner/_work/beam/beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:89: error: [argument] incompatible argument for parameter t of PCollection.apply.
744
? DataflowGroupByKey.createWithAllowDuplicates()
```
--
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 a customized DataflowGroupByKey primitive transform to plumb Redistribute transform's property [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #31490:
URL: https://github.com/apache/beam/pull/31490#discussion_r1635130841
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/RedistributeByKeyOverrideFactory.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * 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.runners.dataflow;
+
+import java.util.Collections;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory;
+import org.apache.beam.sdk.runners.PTransformOverrideFactory.PTransformReplacement;
+import org.apache.beam.sdk.transforms.DataflowGroupByKey;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.Element;
+import org.apache.beam.sdk.transforms.DoFn.OutputReceiver;
+import org.apache.beam.sdk.transforms.DoFn.ProcessElement;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Redistribute.RedistributeByKey;
+import org.apache.beam.sdk.transforms.Reify;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.util.IdentityWindowFn;
+import org.apache.beam.sdk.util.construction.PTransformReplacements;
+import org.apache.beam.sdk.util.construction.SingleInputOutputOverrideFactory;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.ValueInSingleWindow;
+import org.apache.beam.sdk.values.WindowingStrategy;
+import org.joda.time.Duration;
+
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
Can you paste the whole message in a code block to make sure I am reading right? It looks like you are passing something where the items in the iterable are `@nullable ValueInSingleWindow` when it requires `@nonnull ValueInSingleWindow` so that would be an actual error.
--
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