You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/12/06 13:31:31 UTC
[GitHub] [beam] mosche opened a new pull request, #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
mosche opened a new pull request, #24543:
URL: https://github.com/apache/beam/pull/24543
When investigating the performance of TPCDS query 83, I noticed that the query plan contains fairly heavy weight tasks. These trigger a Spark warning when the tasks get broadcasted because their serialized size (using Java serialization) is above 1MB.
This is both a cleanup of the ParDo translation, but also reduces the serialized size of tasks that include many ParDos.
- Use subtypes of `DoFnMapPartitionsFactory` to handle the single output and multi-output case, rather than relying on a closure that has to be serialized.
- Use Java default collections rather than immutable Guava collections (Why does this matter for serialization???).
- Use `Collections.emptyList` and similar where applicable.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] 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://beam.apache.org/contribute/get-started-contributing/#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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1357232658
Thanks a lot @echauchot , I've addressed all your 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
[GitHub] [beam] github-actions[bot] commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1339396253
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1052053003
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
Review Comment:
:+1
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
Review Comment:
:+1:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1351349498
Kind ping @aromanenko-dev @echauchot
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1050531362
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java:
##########
@@ -115,40 +108,27 @@ public boolean canTranslate(ParDo.MultiOutput<InputT, OutputT> transform) {
@Override
public void translate(ParDo.MultiOutput<InputT, OutputT> transform, Context cxt)
throws IOException {
- String stepName = cxt.getCurrentTransform().getFullName();
-
- TupleTag<OutputT> mainOutputTag = transform.getMainOutputTag();
-
- DoFnSchemaInformation doFnSchema =
- ParDoTranslation.getSchemaInformation(cxt.getCurrentTransform());
PCollection<InputT> input = (PCollection<InputT>) cxt.getInput();
- Map<String, PCollectionView<?>> sideInputs = transform.getSideInputs();
Map<TupleTag<?>, PCollection<?>> outputs = cxt.getOutputs();
- DoFnMapPartitionsFactory<InputT, OutputT> factory =
Review Comment:
I like this sub-types per single output and multi output + the way you managed column indexes.
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
Review Comment:
`? super IntT` is only to comply with the return type of `AppliedPTransform#getTransform()` now that you get the DoFn from the AppliedPTransform right ?
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -136,15 +211,15 @@
}
@Override
- protected OutputT computeNext() {
+ protected OutT computeNext() {
Review Comment:
it was already clear but now if a user asks the difference between a Beam bundle and a spark partition, I could answer "none" and point to this code that invokes startBundle/finishBundle with each spark partition.
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
Review Comment:
what is the aim of adding `extends Object to OutT ?`
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
Review Comment:
instead of enclosing the Fun1 you put it one level up by implementing it. One more reduce of the closure !
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
Review Comment:
I know it is not your naming but please rename to DoFnPartitionIterator for clarity
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
Review Comment:
OutT and FnOutT can be different ?
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
private final DoFnSchemaInformation doFnSchema;
private final Supplier<PipelineOptions> options;
-
private final Coder<InT> coder;
private final WindowingStrategy<?, ?> windowingStrategy;
- private final TupleTag<OutT> mainOutput;
+ private final TupleTag<FnOutT> mainOutput;
private final List<TupleTag<?>> additionalOutputs;
private final Map<TupleTag<?>, Coder<?>> outputCoders;
-
private final Map<String, PCollectionView<?>> sideInputs;
private final SideInputReader sideInputReader;
- DoFnMapPartitionsFactory(
- String stepName,
- DoFn<InT, OutT> doFn,
- DoFnSchemaInformation doFnSchema,
+ private DoFnMapPartitionsFactory(
+ AppliedPTransform<PCollection<? extends InT>, ?, MultiOutput<InT, FnOutT>> appliedPT,
Review Comment:
closure reduced a lot !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1354383467
@mosche BTW Flink uses https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/ClosureCleaner.java for similar purpose that could be worth it taking a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1051898620
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
Review Comment:
> OutT and FnOutT can be different ?
`OutT` is the output type of the partition iterator, in case of multiple outputs that's the tupled encoding. `FnOutT` is the internal output type of the `DoFn`.
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
Review Comment:
> OutT and FnOutT can be different ?
`OutT` is the output type of the partition iterator, in case of multiple outputs that's the tupled encoding. `FnOutT` is the internal output type of the `DoFn`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1352867517
@mosche taking a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1051898910
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
+ implements Function1<Iterator<WindowedValue<InT>>, Iterator<OutT>>, Serializable {
private final String stepName;
-
- private final DoFn<InT, OutT> doFn;
+ private final DoFn<? super InT, FnOutT> doFn;
Review Comment:
`? super IntT` is a left over, removed 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1357444108
Run Spark StructuredStreaming ValidatesRunner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1051883181
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
+ * {@link DoFnRunners.OutputManager}.
*/
-class DoFnMapPartitionsFactory<InT, OutT> implements Serializable {
+abstract class DoFnMapPartitionsFactory<InT, FnOutT, OutT extends @NonNull Object>
Review Comment:
That's to make checker happy, it requires `OutT` to be `NonNull`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1339395123
R: @aromanenko-dev
R: @echauchot
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot commented on PR #24543:
URL: https://github.com/apache/beam/pull/24543#issuecomment-1352925423
Run Spark StructuredStreaming ValidatesRunner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] mosche commented on a diff in pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
mosche commented on code in PR #24543:
URL: https://github.com/apache/beam/pull/24543#discussion_r1051899449
##########
runners/spark/3/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/DoFnMapPartitionsFactory.java:
##########
@@ -17,91 +17,171 @@
*/
package org.apache.beam.runners.spark.structuredstreaming.translation.batch;
-import static java.util.stream.Collectors.toCollection;
-import static java.util.stream.Collectors.toMap;
import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.scalaIterator;
-import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists.newArrayListWithCapacity;
+import static org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.tuple;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import java.io.Serializable;
import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Deque;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.apache.beam.runners.core.DoFnRunner;
import org.apache.beam.runners.core.DoFnRunners;
-import org.apache.beam.runners.core.DoFnRunners.OutputManager;
import org.apache.beam.runners.core.SideInputReader;
+import org.apache.beam.runners.core.construction.ParDoTranslation;
import org.apache.beam.runners.spark.structuredstreaming.metrics.MetricsAccumulator;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.CachedSideInputReader;
import org.apache.beam.runners.spark.structuredstreaming.translation.batch.functions.NoOpStepContext;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun1;
-import org.apache.beam.runners.spark.structuredstreaming.translation.utils.ScalaInterop.Fun2;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFnSchemaInformation;
+import org.apache.beam.sdk.transforms.ParDo.MultiOutput;
import org.apache.beam.sdk.transforms.reflect.DoFnInvokers;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.AbstractIterator;
-import org.apache.spark.api.java.function.MapPartitionsFunction;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.checkerframework.checker.nullness.qual.NonNull;
+import scala.Function1;
+import scala.Tuple2;
import scala.collection.Iterator;
/**
- * Encapsulates a {@link DoFn} inside a Spark {@link
- * org.apache.spark.api.java.function.MapPartitionsFunction}.
+ * Abstract factory to create a {@link DoFnPartitionIt DoFn partition iterator} using a customizable
Review Comment:
👍 will do
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot merged pull request #24543: [Spark Dataset runner] Reduce binary size of Java serialized task related for ParDo translation
Posted by GitBox <gi...@apache.org>.
echauchot merged PR #24543:
URL: https://github.com/apache/beam/pull/24543
--
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