[BEAM-647] Fault-tolerant sideInputs via Broadcast variables #1624

kobisalant · 2016-12-15T17:51:51Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

asfbot · 2016-12-15T17:54:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/5952/
--none--

amitsela · 2016-12-17T15:29:09Z

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java

@@ -324,7 +327,7 @@ public void visitPrimitiveTransform(TransformHierarchy.Node node) {
      LOG.info("Evaluating {}", transform);
      AppliedPTransform<?, ?, ?> appliedTransform = node.toAppliedPTransform();
      ctxt.setCurrentTransform(appliedTransform);
-      evaluator.evaluate(transform, ctxt);
+      evaluator.evaluate(transform, ctxt, pview);


Why pass the SparkPCollectionView when the PipelineVisitor traverses ? this only happens once, on first run. I'm also pretty sure this never happens again (such as on resume from checkpoint)

amitsela · 2016-12-17T15:30:50Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+	// Driver only
+	BroadcastHelper getPCollectionView(PCollectionView<?> view, JavaSparkContext context) {
+		// initialize broadcastHelperMap if needed
+		if (broadcastHelperMap == null) {


If you use double-lock synchronization broadcastHelperMap should be volatile.

Right, fixed

amitsela · 2016-12-17T15:33:11Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.spark.api.java.JavaSparkContext;
+
+public class SparkPCollectionView implements Serializable {


This should probably host some kind of thread-safe, lazy, Singleton. this might also save the trouble of passing this through the TransformEvaluator.

Fix comments by Amit + rebase from master + checkstyle

asfbot · 2016-12-18T16:04:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6061/
--none--

kobisalant · 2016-12-19T11:41:08Z

R. @amitsela

asfbot · 2016-12-19T12:06:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6068/
--none--

amitsela · 2017-01-01T17:31:18Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

@@ -129,10 +128,6 @@ public void putDataset(PValue pvalue, Dataset dataset) {
    datasets.put((PValue) getOutput(transform), new UnboundedDataset<>(values, jssc, coder));
  }

-  void putPView(PValue view, Iterable<? extends WindowedValue<?>> value) {


Why not keep put/get encapsulated here ? saves the ugly context.getPviews().putPViews()

Will be fixed

amitsela · 2017-01-01T17:33:55Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

@@ -149,10 +144,6 @@ public Dataset borrowDataset(PValue pvalue) {
    return dataset;
  }

-  <T> Iterable<? extends WindowedValue<?>> getPCollectionView(PCollectionView<T> view) {


Same as above, why not get the appropriate BroadcastHelper ?

It remains the same as before the TranslationUtils.getSideInputs gets a new parameter "pviews" and is called inside the lambda, everything else is the same

amitsela · 2017-01-01T17:34:21Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+


Please remove empty line.

will be fixed

asfbot · 2017-01-02T13:10:36Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6361/

Build result: FAILURE

[...truncated 11169 lines...] at hudson.remoting.UserRequest.perform(UserRequest.java:153) at hudson.remoting.UserRequest.perform(UserRequest.java:50) at hudson.remoting.Request$2.run(Request.java:332) at hudson.remoting.InterceptingExecutorService$1.call(InterceptingExecutorService.java:68) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)Caused by: org.apache.maven.plugin.MojoFailureException: You have 4 Checkstyle violations. at org.apache.maven.plugin.checkstyle.CheckstyleViolationCheckMojo.execute(CheckstyleViolationCheckMojo.java:588) at org.apache.maven.plugin.DefaultBuildPluginManager.executeMojo(DefaultBuildPluginManager.java:134) at org.apache.maven.lifecycle.internal.MojoExecutor.execute(MojoExecutor.java:208) ... 31 more2017-01-02T13:10:32.256 [ERROR] 2017-01-02T13:10:32.256 [ERROR] Re-run Maven using the -X switch to enable full debug logging.2017-01-02T13:10:32.256 [ERROR] 2017-01-02T13:10:32.257 [ERROR] For more information about the errors and possible solutions, please read the following articles:2017-01-02T13:10:32.257 [ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoFailureException2017-01-02T13:10:32.257 [ERROR] 2017-01-02T13:10:32.257 [ERROR] After correcting the problems, you can resume the build with the command2017-01-02T13:10:32.257 [ERROR] mvn -rf :beam-runners-sparkchannel stoppedSetting status of d0815ea to FAILURE with url https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6361/ and message: 'Build finished. 'Using context: Jenkins: Maven clean install
--none--

asfbot · 2017-01-02T13:52:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6364/
--none--

asfbot · 2017-01-02T15:23:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6366/

Failed Tests: 1

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-apex: 1

org.apache.beam.runners.apex.translation.ParDoBoundTranslatorTest.testAssertionFailure

--none--

amitsela · 2017-01-03T08:14:43Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+    // Holds the Actual data of the views in serialize form
+    private Map<PCollectionView<?>,
+        Tuple2<byte[], Coder<Iterable<WindowedValue<?>>>>> pviews =
+            Collections.synchronizedMap(


I don't think there's a real need for a synchronized map.

amitsela · 2017-01-03T08:20:17Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+    private transient volatile Map<PCollectionView<?>, BroadcastHelper> broadcastHelperMap = null;
+
+    // Holds the Actual data of the views in serialize form
+    private Map<PCollectionView<?>,


Since you rely on SparkPCollectionView to be serialized as part of the DStream (checkpointing), and this Map to be persisted along, it should be noted that this is sub-optimal since all side inputs would be persisted for every DoFn they appear in.

amitsela · 2017-01-03T08:24:32Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+                    Tuple2<byte[], Coder<Iterable<WindowedValue<?>>>>>());
+
+    // Driver only - during evaluation stage
+    void putPView(PCollectionView<?> view,


Nit: indent parameters.

throughout this class.

amitsela · 2017-01-03T08:27:25Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+            }
+        } else if (overwrite) {
+            synchronized (SparkPCollectionView.class) {
+                helper.unpersist(context);


Can comment that unpersist can be blocking.. if we'd want to consider this in the future.

amitsela · 2017-01-03T08:28:21Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+        return helper;
+    }
+
+    private BroadcastHelper getBroadcastHelper(PCollectionView<?> view, JavaSparkContext context) {


this is more of createBroadcastHelper

amitsela · 2017-01-03T08:30:08Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

+   * @param value - Actual value of the view
+   * @param coder - Coder of the value
+   */
+  public void putPView(PCollectionView<?> view,


indent first parameter as well.

First parameter should still be aligned with others.
Also, naming should match the getter - either both with capital V or not: getPViews/putPView or getPviews/putPviews.

amitsela · 2017-01-03T08:39:30Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TranslationUtils.java

@@ -192,22 +192,30 @@ public Boolean call(Tuple2<TupleTag<V>, WindowedValue<?>> input) {
   * @return a map of tagged {@link BroadcastHelper}s and their {@link WindowingStrategy}.
   */
  public static Map<TupleTag<?>, KV<WindowingStrategy<?, ?>, BroadcastHelper<?>>>
-      getSideInputs(List<PCollectionView<?>> views, EvaluationContext context) {
+  getSideInputs(List<PCollectionView<?>> views, EvaluationContext context) {


This one seems unused.

used throughout TransformTranslator, prefer to keep it as is

It appears that this can be package-private, which is also nicer since it has an overriding implementation.

amitsela · 2017-01-03T08:51:51Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/BroadcastHelper.java

-      return new DirectBroadcastHelper<>(value);
-    }
-    return new CodedBroadcastHelper<>(value, coder);
+  public static <T> BroadcastHelper<T> create(byte[] bytes, Coder<T> coder) {


Can get rid of the abstraction.
I don't see a reason to use "direct" translation since Beam heavily relies on Coders anyway.

amitsela · 2017-01-03T15:03:24Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+        Coder<Iterable<WindowedValue<?>>> coder,
+        JavaSparkContext context) {
+
+        pviews.put(view, new Tuple2(CoderHelpers.toByteArray(value, coder), coder));


Unchecked - missing "diamonds" - new Tuple2<>(....)

amitsela · 2017-01-03T15:04:38Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/BroadcastHelper.java

+      this.bcast = jsc.broadcast(bytes);
+    }
+
+   public void unpersist(JavaSparkContext jsc) {


unused method parameter.

asfbot · 2017-01-03T15:18:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6374/
--none--

asfbot · 2017-01-03T15:48:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6375/
--none--

asfbot · 2017-01-04T10:40:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6394/
--none--

amitsela · 2017-01-04T11:35:27Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+
+        pviews.put(view, new Tuple2<>(CoderHelpers.toByteArray(value, coder), coder));
+        // overwrite/create broadcast
+        getPCollectionView(view, context, true);


Thought: should broadcast be eager ? maybe this line could be removed ?
Should putPView simply set the view in pviews while the actual broadcast will happen on demand ?

I had some issues with this approach, the Write operation is a composite transform that passes side inputs between the transformation, can't remember exactly what was the problem. We can try it and see how it goes. There were many changes since the lazy version. We need to think also of updating the broadcast variable, I think it should work lazily but...

Like I said, mostly a thought.. If it's not trivial, this could be considered for future work in a separate ticket if it would benefit.

amitsela · 2017-01-04T11:38:53Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+public class SparkPCollectionView implements Serializable {
+
+    // Holds the view --> broadcast mapping. Transient so it will be null from resume
+    private transient volatile Map<PCollectionView<?>, BroadcastHelper> broadcastHelperMap = null;


Won't it be simpler to instantiate this map eagerly ? avoid the double-checked locking ?

I think we can do it eagerly, Let me try it out

On second thought, we will need to handle it in the Deserialize. Do you think it will be easier to add resolve methods to the class? it may be dependant on the serialization framework (kryo, java, etc)

Agree, keep it lazy as it is.

amitsela · 2017-01-04T11:47:27Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/BroadcastHelper.java

@@ -31,84 +30,42 @@
 /**
 * Broadcast helper.
 */
-public abstract class BroadcastHelper<T> implements Serializable {
+public class BroadcastHelper<T> implements Serializable {


This class seems to suffer some alignment issues, probably from removing the abstract and inner implementations.

amitsela · 2017-01-04T11:49:33Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/SparkSideInputReader.java

@@ -61,6 +61,7 @@ public SparkSideInputReader(Map<TupleTag<?>, KV<WindowingStrategy<?, ?>, Broadca
    //--- match the appropriate sideInput window.
    // a tag will point to all matching sideInputs, that is all windows.
    // now that we've obtained the appropriate sideInputWindow, all that's left is to filter by it.
+    @SuppressWarnings("unchecked")


Which warning do you suppress here ?
I don't see anything in my IDE..

Unchecked cast: 'java.lang.Object' to 'java.lang.Iterable<org.apache.beam.sdk.util.WindowedValue<?>>' less... (⌘F1)
Signals places where an unchecked warning is issued by the compiler, for example:

void f(HashMap map) {
map.put("key", "value");
}

Let's remove this for now.

asfbot · 2017-01-04T13:28:26Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6395/
--none--

asfbot · 2017-01-05T08:05:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6425/
--none--

amitsela

Mostly Nit picking, almost there.

amitsela · 2017-01-10T13:23:27Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

@@ -201,6 +193,26 @@ public void computeOutputs() {
    return Iterables.transform(windowedValues, WindowingHelpers.<T>unwindowValueFunction());
  }

+  /**
+   * Retruns the current views creates in the pipepline.


Typo: Return the current views crated by the pipeline.
Also a newline before @return.

amitsela · 2017-01-10T13:25:16Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

+  }
+
+  /**
+   * Adds/Replaces a view to the current views creates in the pipepline.


Same here, newline between Adds/Replaces... and @param.

amitsela · 2017-01-10T13:27:58Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/EvaluationContext.java

+   * @param value - Actual value of the view
+   * @param coder - Coder of the value
+   */
+  public void putPView(PCollectionView<?> view,


First parameter should still be aligned with others.
Also, naming should match the getter - either both with capital V or not: getPViews/putPView or getPviews/putPviews.

amitsela · 2017-01-10T13:29:02Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+public class SparkPCollectionView implements Serializable {
+
+    // Holds the view --> broadcast mapping. Transient so it will be null from resume
+    private transient volatile Map<PCollectionView<?>, BroadcastHelper> broadcastHelperMap = null;


Agree, keep it lazy as it is.

amitsela · 2017-01-10T14:02:13Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkPCollectionView.java

+        PCollectionView<?> view,
+        Iterable<WindowedValue<?>> value,
+        Coder<Iterable<WindowedValue<?>>> coder,
+        JavaSparkContext context) {


context is unused.

amitsela · 2017-01-10T14:09:29Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/TranslationUtils.java

  public static Map<TupleTag<?>, KV<WindowingStrategy<?, ?>, BroadcastHelper<?>>>
-      getSideInputs(List<PCollectionView<?>> views, EvaluationContext context) {
+  getSideInputs(List<PCollectionView<?>> views, JavaSparkContext context,


Indent parameters.

amitsela · 2017-01-10T14:12:25Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/BroadcastHelper.java

@@ -31,97 +30,53 @@
 /**
 * Broadcast helper.
 */
-public abstract class BroadcastHelper<T> implements Serializable {
+public class BroadcastHelper<T> implements Serializable {


Maybe this should not be called BroadcastHelper anymore... SideInputBroadcast ?

The purpose of this class was not changed, only the constructor is from bytes and not T. I think there is no reason to change it. it is not tightly coupled to side input

amitsela · 2017-01-10T14:13:25Z

runners/spark/src/main/java/org/apache/beam/runners/spark/util/SparkSideInputReader.java

@@ -61,6 +61,7 @@ public SparkSideInputReader(Map<TupleTag<?>, KV<WindowingStrategy<?, ?>, Broadca
    //--- match the appropriate sideInput window.
    // a tag will point to all matching sideInputs, that is all windows.
    // now that we've obtained the appropriate sideInputWindow, all that's left is to filter by it.
+    @SuppressWarnings("unchecked")


Let's remove this for now.

amitsela · 2017-01-10T14:14:31Z

...a/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java

+          .apply(ParDo.of(new DoFn<KV<String, String>, KV<String, String>>() {
+               @ProcessElement
+               public void process(ProcessContext c) {
+


Remove empty line.

amitsela · 2017-01-10T14:16:46Z

...a/org/apache/beam/runners/spark/translation/streaming/ResumeFromCheckpointStreamingTest.java

+               @ProcessElement
+               public void process(ProcessContext c) {
+
+                  // Check side input is passed correctly


"... also after resuming from checkpoint."

asfbot · 2017-01-10T16:15:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6481/
--none--

amitsela · 2017-01-10T16:44:28Z

@kobisalant would you mind checking the proper checkboxes in the PR template ? thanks!

amitsela · 2017-01-10T16:50:11Z

LGTM. Thanks @kobisalant !

amitsela reviewed Dec 17, 2016

View reviewed changes

[BEAM-647] Fault-tolerant sideInputs via Broadcast variables

0ea9f1a

Fix comments by Amit + rebase from master + checkstyle

kobisalant force-pushed the BEAM-647-Fault-tolerant-sideInputs branch from 228c613 to 0ea9f1a Compare December 18, 2016 15:39

Reformat + add unpersist on push

810eafb

amitsela reviewed Jan 1, 2017

View reviewed changes

Fix Amit's comments

d0815ea

Fix Amit's comments

a1550ec

fix spaces

c0548d2

amitsela reviewed Jan 3, 2017

View reviewed changes

fix Amit's comments

41022ba

amitsela reviewed Jan 3, 2017

View reviewed changes

fix Amit's comments

e6575c2

fix Warnings

7e7b34a

amitsela reviewed Jan 4, 2017

View reviewed changes

fix Warnings

aa9c4ba

Create broadcast lazily

8365493

amitsela reviewed Jan 10, 2017

View reviewed changes

Fix Amit's comments + rename BroadcastHelper to SideInputBroadcast

0352803

asfgit closed this in c1b7f86 Jan 10, 2017

[BEAM-647] Fault-tolerant sideInputs via Broadcast variables #1624

[BEAM-647] Fault-tolerant sideInputs via Broadcast variables #1624

Conversation

kobisalant commented Dec 15, 2016

asfbot commented Dec 15, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Dec 18, 2016

kobisalant commented Dec 19, 2016

asfbot commented Dec 19, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 2, 2017

Build result: FAILURE

asfbot commented Jan 2, 2017

asfbot commented Jan 2, 2017

Failed Tests: 1

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-apex: 1

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amitsela Jan 3, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 3, 2017

asfbot commented Jan 3, 2017

asfbot commented Jan 4, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 4, 2017

asfbot commented Jan 5, 2017

amitsela left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 10, 2017

amitsela Jan 3, 2017 •

edited

Loading