This is an automated email from the ASF dual-hosted git repository. iemejia pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push: new 2d9bfd5 [BEAM-6966] Spark portable runner: get PAssert working new e413098 Merge pull request #8323: [BEAM-6966] Spark portable runner: get PAssert working 2d9bfd5 is described below commit 2d9bfd54737221c6472632cc9d4a49ef1a149f25 Author: Kyle Weaver <kcwea...@google.com> AuthorDate: Tue Apr 2 14:45:40 2019 -0700 [BEAM-6966] Spark portable runner: get PAssert working --- .../SparkBatchPortablePipelineTranslator.java | 9 +++++++++ .../runners/spark/SparkPortableExecutionTest.java | 21 +++++++++++++++------ 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkBatchPortablePipelineTranslator.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkBatchPortablePipelineTranslator.java index 2ad0669..12f4ea9 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkBatchPortablePipelineTranslator.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkBatchPortablePipelineTranslator.java @@ -215,6 +215,15 @@ public class SparkBatchPortablePipelineTranslator { staged.flatMap(new SparkExecutableStageExtractionFunction<>(outputMap.get(outputId))); context.pushDataset(outputId, new BoundedDataset<>(outputRdd)); } + if (outputs.isEmpty()) { + // After pipeline translation, we traverse the set of unconsumed PCollections and add a + // no-op sink to each to make sure they are materialized by Spark. However, some SDK-executed + // stages have no runner-visible output after fusion. We handle this case by adding a sink + // here. + JavaRDD<WindowedValue<OutputT>> outputRdd = + staged.flatMap((rawUnionValue) -> Collections.emptyIterator()); + context.pushDataset("EmptyOutputSink", new BoundedDataset<>(outputRdd)); + } } /** diff --git a/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkPortableExecutionTest.java b/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkPortableExecutionTest.java index 38bdd1f..d7d3428 100644 --- a/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkPortableExecutionTest.java +++ b/runners/spark/src/test/java/org/apache/beam/runners/spark/SparkPortableExecutionTest.java @@ -18,11 +18,13 @@ package org.apache.beam.runners.spark; import java.io.Serializable; +import java.util.Collections; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.apache.beam.model.jobmanagement.v1.JobApi.JobState.Enum; import org.apache.beam.model.pipeline.v1.RunnerApi; import org.apache.beam.runners.core.construction.Environments; +import org.apache.beam.runners.core.construction.JavaReadViaImpulse; import org.apache.beam.runners.core.construction.PipelineTranslation; import org.apache.beam.runners.fnexecution.jobsubmission.JobInvocation; import org.apache.beam.sdk.Pipeline; @@ -30,6 +32,7 @@ import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PipelineOptionsFactory; import org.apache.beam.sdk.options.PortablePipelineOptions; import org.apache.beam.sdk.testing.CrashingRunner; +import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.GroupByKey; import org.apache.beam.sdk.transforms.Impulse; @@ -39,6 +42,7 @@ import org.apache.beam.sdk.transforms.WithKeys; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.sdk.values.PCollectionView; +import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v20_0.com.google.common.util.concurrent.ListeningExecutorService; import org.apache.beam.vendor.guava.v20_0.com.google.common.util.concurrent.MoreExecutors; import org.junit.AfterClass; @@ -83,15 +87,15 @@ public class SparkPortableExecutionTest implements Serializable { Pipeline p = Pipeline.create(options); - final PCollectionView<Integer> view = + final PCollectionView<Long> view = p.apply("impulse23", Impulse.create()) .apply( "create23", ParDo.of( - new DoFn<byte[], Integer>() { + new DoFn<byte[], Long>() { @ProcessElement public void process(ProcessContext context) { - context.output(23); + context.output(23L); } })) .apply(View.asSingleton()); @@ -127,8 +131,7 @@ public class SparkPortableExecutionTest implements Serializable { new DoFn<KV<String, Iterable<Long>>, KV<String, Long>>() { @ProcessElement public void process(ProcessContext context) { - LOG.info("Side input: {}", context.sideInput(view)); - LOG.info("Output element: {}", context.element()); + context.output(KV.of("bar", context.sideInput(view))); for (Long i : context.element().getValue()) { context.output(KV.of(context.element().getKey(), i)); } @@ -138,7 +141,13 @@ public class SparkPortableExecutionTest implements Serializable { // Second GBK forces the output to be materialized .apply("gbk", GroupByKey.create()); - // TODO Get PAssert working to test pipeline result + PAssert.that(result) + .containsInAnyOrder( + KV.of("foo", ImmutableList.of(4L, 3L, 3L)), KV.of("bar", ImmutableList.of(23L))); + + // This is line below required to convert the PAssert's read to an impulse, which is expected + // by the GreedyPipelineFuser. + p.replaceAll(Collections.singletonList(JavaReadViaImpulse.boundedOverride())); RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);