[BEAM-920] Add support for Watermarks in the Spark runner. by amitsela · Pull Request #1987 · apache/beam

amitsela · 2017-02-12T18:11:15Z

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.

amitsela · 2017-02-12T18:19:42Z

This PR is the first of 2-3 PRs that will complete BEAM-920 to support the Beam model for streaming.
I'm trying to break to as many separate PRs as possible to make reviews simpler and focused, though it's not always easy since a lot of things are entangled here.

This PR adds support to Watermarks via a global Broadcast to workers, that keeps updating every batch (asynchronously), and also allows the runner's UnboundedDatasets (representation of UNBOUNDED PCollections) to track their creating sources so in case of a flattened stream they could access the relevant WMs.

amitsela · 2017-02-12T18:20:51Z

R: @kennknowles
CC: @staslev

jbonofre · 2017-02-12T18:27:25Z

CC : @jbonofre

coveralls · 2017-02-12T18:43:00Z

Coverage decreased (-0.03%) to 69.679% when pulling e3593b5 on amitsela:watermarks into 5fe11a2 on apache:master.

asfbot · 2017-02-12T18:43:28Z

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

kennknowles · 2017-02-17T22:41:53Z

runners/spark/src/test/java/org/apache/beam/runners/spark/WatermarkTest.java

+
+    p.run().waitUntilFinish(Duration.millis(options.getBatchIntervalMillis()).multipliedBy(3));
+
+    System.out.println(WatermarksDoFn.strings);


System.out.println

kennknowles · 2017-02-17T22:42:22Z

runners/spark/src/test/java/org/apache/beam/runners/spark/WatermarkTest.java

+
+    @ProcessElement
+    public void processElement(ProcessContext c) {
+      if (GlobalWatermarkHolder.get() == null


This should be able to be replaced by TestStream + triggers, right?

Yes, next PR, lining it up as we speak.

kennknowles · 2017-02-17T22:43:47Z

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

+        Queue<SparkWatermarks> timesQueue = en.getValue();
+
+        // current state, if exists.
+        Instant currentLowWatermark = BoundedWindow.TIMESTAMP_MIN_VALUE;


This logic seems about right. I admit that I have forgotten some of the details of our deep dive on watermarks.

I've updated the design doc: https://docs.google.com/document/d/1bOu0-g7Z1LVG0kYEQ38GDWaFjFl45u-2aZV87RmlLKI/edit?usp=sharing

kennknowles · 2017-02-17T22:45:14Z

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

+      timesQueue = new ConcurrentLinkedQueue<>();
+    }
+    timesQueue.offer(sparkWatermarks);
+    sourceTimes.put(sourceId, timesQueue);


I see that the watermarks for unbounded sources are tracked here. What I don't quite follow is the behavior in fan-in situations where the watermark is the min of the inputs.

~~So in cases such as Flatten where the WM is the min{WM1, WM2} it is up to whoever uses WMs to deduce that (such as GABW implementation).~~
This would be handled by SparkTimerInternals.
This WM mechanism only reflects the current-batch lo/hi WMs per source. Adding to that the information on upstream/origin sources to a PCollection is supposed to provide all necessary information to find the right WM for the computation, so for example a GABW following a Flatten would have to choose the Min{WM...}.
This is implemented in the next PR as well since it's related to triggers and generally a part of GABW implementation.

See here in the branch I'm preparing for the next PR: https://github.com/amitsela/beam/blob/triggers-and-panes/runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkTimerInternals.java#L66

amitsela · 2017-02-18T10:18:56Z

@kennknowles that's sort of the trade-off between gradual PRing with (fairly) small changes or one huge PR.
I've added some tests to make this as complete as possible but clearly it'll make more sense after the next PR.
I'll finish the polishing there and add a link to the branch here so you'd get a sneak-preview 😉

…ould be revisited there.

amitsela · 2017-02-18T21:22:56Z

Rebased and removed leftover printouts.

asfbot · 2017-02-18T21:40:50Z

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

Build result: FAILURE

[...truncated 12098 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 1 Checkstyle violation. 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-02-18T21:40:39.238 [ERROR] 2017-02-18T21:40:39.238 [ERROR] Re-run Maven using the -X switch to enable full debug logging.2017-02-18T21:40:39.238 [ERROR] 2017-02-18T21:40:39.238 [ERROR] For more information about the errors and possible solutions, please read the following articles:2017-02-18T21:40:39.238 [ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoFailureException2017-02-18T21:40:39.239 [ERROR] 2017-02-18T21:40:39.239 [ERROR] After correcting the problems, you can resume the build with the command2017-02-18T21:40:39.239 [ERROR] mvn -rf :beam-runners-sparkchannel stoppedSetting status of c2415a0 to FAILURE with url https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/7583/ and message: 'Build finished. 'Using context: Jenkins: Maven clean install
--none--

amitsela · 2017-02-19T06:50:59Z

retest this please.

coveralls · 2017-02-19T07:26:33Z

Coverage decreased (-0.03%) to 69.166% when pulling 7463d8c on amitsela:watermarks into 01de255 on apache:master.

asfbot · 2017-02-19T07:27:19Z

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

Failed Tests: 1

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

org.apache.beam.runners.spark.WatermarkTest.testInDoFn

--none--

amitsela · 2017-02-19T07:44:47Z

retest this please.

coveralls · 2017-02-19T08:15:26Z

Coverage decreased (-0.03%) to 69.166% when pulling 7463d8c on amitsela:watermarks into 01de255 on apache:master.

asfbot · 2017-02-19T08:16:10Z

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

amitsela · 2017-02-19T11:04:30Z

Test flake was weird, but I don't see a point in investing too much here since this test specifically is to be removed in the next PR where Watermarks are part of "TestStream-like" test pipelines.

kennknowles

OK, LGTM. I understand that these things are going to have more changes as they develop, anyhow, so let's keep momentum. Sorry it took so long to grok this.

amitsela · 2017-02-20T09:07:27Z

Added a fix to clear the state of singletons BEFORE pipeline execution via TestSparkRunner - this broke MetricsTest in post commit (I ran locally before merging). Simply moving this before executing the pipeline.
Merging when tests here pass and post commits pass locally for me.

coveralls · 2017-02-20T09:40:29Z

Coverage decreased (-0.02%) to 69.173% when pulling 6ba7ed6 on amitsela:watermarks into 01de255 on apache:master.

asfbot · 2017-02-20T09:47:00Z

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

kennknowles reviewed Feb 17, 2017

View reviewed changes

Sela added 11 commits February 18, 2017 23:15

A global Watermark holder to update and broadcast to workers.

3f39f52

Ingest the input watermarks into the GlobalWatermarkHolder.

fa1af9e

Advance watermarks onBatchCompleted hook.

1b76632

Handle QueuedStream (for testing) and track sources upstream.

6fdee4b

Refactor according to changes.

f336168

Exopse RegexMatcher as part of the SDK.

3c4e064

Watermark tests.

d85099d

Streaming sources tracking test.

0cd4d11

This relied on a wrong functionality as descrined in BEAM-1444 and sh…

8c52909

…ould be revisited there.

fixup! licenese.

f48f2e4

fixup! remove leftover printouts.

c2415a0

amitsela force-pushed the watermarks branch from e3593b5 to c2415a0 Compare February 18, 2017 21:22

fixup! import order.

7463d8c

kennknowles approved these changes Feb 20, 2017

View reviewed changes

fixup! clear state of static singletons before each test.

6ba7ed6

asfgit closed this in aa45ccb Feb 20, 2017


		p.run().waitUntilFinish(Duration.millis(options.getBatchIntervalMillis()).multipliedBy(3));

		System.out.println(WatermarksDoFn.strings);

Conversation

amitsela commented Feb 12, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

amitsela commented Feb 12, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

amitsela commented Feb 12, 2017

Uh oh!

jbonofre commented Feb 12, 2017

Uh oh!

coveralls commented Feb 12, 2017

Uh oh!

asfbot commented Feb 12, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amitsela Feb 18, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amitsela commented Feb 18, 2017

Uh oh!

amitsela commented Feb 18, 2017

Uh oh!

asfbot commented Feb 18, 2017

Build result: FAILURE

Uh oh!

amitsela commented Feb 19, 2017

Uh oh!

coveralls commented Feb 19, 2017

Uh oh!

asfbot commented Feb 19, 2017

Failed Tests: 1

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

Uh oh!

amitsela commented Feb 19, 2017

Uh oh!

coveralls commented Feb 19, 2017

Uh oh!

asfbot commented Feb 19, 2017

Uh oh!

amitsela commented Feb 19, 2017

Uh oh!

kennknowles left a comment

Choose a reason for hiding this comment

Uh oh!

amitsela commented Feb 20, 2017

Uh oh!

coveralls commented Feb 20, 2017

Uh oh!

asfbot commented Feb 20, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

amitsela commented Feb 12, 2017 •

edited

Loading

amitsela commented Feb 12, 2017 •

edited

Loading

amitsela Feb 18, 2017 •

edited

Loading