You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ec...@apache.org on 2019/07/16 14:53:20 UTC
[beam] branch spark-runner_structured-streaming updated (b62a593 ->
3e6a9e2)
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a change to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git.
from b62a593 Add spark execution plans extended debug messages.
new 9a1e4dc Print number of leaf datasets
new 3e6a9e2 fixup! Add PipelineResults to Spark structured streaming.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../spark/structuredstreaming/translation/TranslationContext.java | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
[beam] 01/02: Print number of leaf datasets
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 9a1e4dcef17c6de1d75d83aa547ea2218d892e95
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Tue Jul 16 15:49:03 2019 +0200
Print number of leaf datasets
---
.../spark/structuredstreaming/translation/TranslationContext.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
index 3ca232c..da610c2 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
@@ -184,6 +184,7 @@ public class TranslationContext {
try {
SparkStructuredStreamingPipelineOptions options =
serializablePipelineOptions.get().as(SparkStructuredStreamingPipelineOptions.class);
+ int datasetIndex = 0;
for (Dataset<?> dataset : leaves) {
if (options.isStreaming()) {
// TODO: deal with Beam Discarding, Accumulating and Accumulating & Retracting outputmodes
@@ -198,6 +199,7 @@ public class TranslationContext {
dataStreamWriter.foreach(new NoOpForeachWriter<>()).start().awaitTermination();
} else {
if (options.getTestMode()) {
+ LOG.debug("**** dataset {}", ++datasetIndex);
dataset.explain(true);
// cannot use dataset.show because dataset schema is binary so it will print binary
// code.
[beam] 02/02: fixup! Add PipelineResults to Spark structured
streaming.
Posted by ec...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch spark-runner_structured-streaming
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 3e6a9e233a18029b887c29c4a53c3fd55bd73597
Author: Ryan Skraba <ry...@skraba.com>
AuthorDate: Tue Jul 16 16:11:58 2019 +0200
fixup! Add PipelineResults to Spark structured streaming.
Materialize rows more aggressively.
---
.../spark/structuredstreaming/translation/TranslationContext.java | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
index da610c2..d7c6a1a 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/TranslationContext.java
@@ -37,6 +37,7 @@ import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.ForeachWriter;
@@ -210,7 +211,9 @@ public class TranslationContext {
} else {
// apply a dummy fn just to apply for each action that will trigger the pipeline run in
// spark
- dataset.foreachPartition(t -> {});
+ // TODO: foreachPartition is too lazy to materialize everything. Check out / compare
+ // foreach or another technique. StructuredStreamingPipelineStateTest can help validate
+ dataset.foreach((ForeachFunction) t -> {});
}
}
}