beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amits...@apache.org
Subject [1/2] beam git commit: Fix caching in the Spark streaming, doing the cache update in the streaming context
Date Sat, 25 Mar 2017 13:23:54 GMT
Repository: beam
Updated Branches:
  refs/heads/master fe2369933 -> 3473055f2


Fix caching in the Spark streaming, doing the cache update in the streaming context

Fix order of call


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/0ec81468
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/0ec81468
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/0ec81468

Branch: refs/heads/master
Commit: 0ec8146848ba3cbb1afb51b5b9919b3ab3db7955
Parents: fe23699
Author: Jean-Baptiste Onofré <jbonofre@apache.org>
Authored: Fri Mar 24 15:27:04 2017 +0100
Committer: Amit Sela <amitsela33@gmail.com>
Committed: Sat Mar 25 16:00:09 2017 +0300

----------------------------------------------------------------------
 .../main/java/org/apache/beam/runners/spark/SparkRunner.java | 8 +-------
 .../streaming/SparkRunnerStreamingContextFactory.java        | 6 ++----
 2 files changed, 3 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/0ec81468/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
index fc5d4af..97532c4 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java
@@ -38,7 +38,6 @@ import org.apache.beam.runners.spark.translation.TransformEvaluator;
 import org.apache.beam.runners.spark.translation.TransformTranslator;
 import org.apache.beam.runners.spark.translation.streaming.Checkpoint.CheckpointDir;
 import org.apache.beam.runners.spark.translation.streaming.SparkRunnerStreamingContextFactory;
-import org.apache.beam.runners.spark.translation.streaming.StreamingTransformTranslator;
 import org.apache.beam.runners.spark.util.GlobalWatermarkHolder.WatermarksListener;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.io.Read;
@@ -163,11 +162,6 @@ public final class SparkRunner extends PipelineRunner<SparkPipelineResult>
{
           JavaStreamingContext.getOrCreate(checkpointDir.getSparkCheckpointDir().toString(),
               contextFactory);
 
-      // update cache candidates
-      translator = new StreamingTransformTranslator.Translator(
-          new TransformTranslator.Translator());
-      updateCacheCandidates(pipeline, translator, contextFactory.getEvaluationContext());
-
       // Checkpoint aggregator/metrics values
       jssc.addStreamingListener(
           new JavaStreamingListenerWrapper(
@@ -269,7 +263,7 @@ public final class SparkRunner extends PipelineRunner<SparkPipelineResult>
{
   /**
    * Evaluator that update/populate the cache candidates.
    */
-  private void updateCacheCandidates(
+  public static void updateCacheCandidates(
       Pipeline pipeline,
       SparkPipelineTranslator translator,
       EvaluationContext evaluationContext) {

http://git-wip-us.apache.org/repos/asf/beam/blob/0ec81468/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
----------------------------------------------------------------------
diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
index c298886..98521e9 100644
--- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
+++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/streaming/SparkRunnerStreamingContextFactory.java
@@ -83,6 +83,8 @@ public class SparkRunnerStreamingContextFactory implements JavaStreamingContextF
     SparkRunner.initAccumulators(options, jsc);
 
     ctxt = new EvaluationContext(jsc, pipeline, jssc);
+    // update cache candidates
+    SparkRunner.updateCacheCandidates(pipeline, translator, ctxt);
     pipeline.traverseTopologically(new SparkRunner.Evaluator(translator, ctxt));
     ctxt.computeOutputs();
 
@@ -91,10 +93,6 @@ public class SparkRunnerStreamingContextFactory implements JavaStreamingContextF
     return jssc;
   }
 
-  public EvaluationContext getEvaluationContext() {
-    return this.ctxt;
-  }
-
   private void checkpoint(JavaStreamingContext jssc) {
     Path rootCheckpointPath = checkpointDir.getRootCheckpointDir();
     Path sparkCheckpointPath = checkpointDir.getSparkCheckpointDir();


Mime
View raw message