BEAM-1438 Auto shard streaming sinks #1952

reuvenlax · 2017-02-08T21:09:44Z

If a Write requests runner-determined sharding, per-bundle sharding is the default but performs poorly in Dataflow's streaming runner. Instead, the runner statically picks a sharding based on the number of workers.

This PR accidentally got closed. Reopening.

R: @jkff

asfbot · 2017-02-08T21:13:31Z

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

coveralls · 2017-02-08T21:58:09Z

Coverage decreased (-0.03%) to 69.585% when pulling 0038d59 on reuvenlax:streaming_auto_shard_write into a25855a on apache:master.

asfbot · 2017-02-08T21:59:57Z

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

dhalperi · 2017-02-21T16:21:21Z

R: @tgroh

tgroh

Tests?

reuvenlax · 2017-02-24T21:13:43Z

@tgroh suggestions on where test should go? I don't see where the tests for similar changes (e.g. replacing PubSub source/sink) are.

tgroh

It's not unreasonable to make the StreamingShardedWrite package-private and have a test for it.

Also, after the submission of #2039, the way this is applied should change - you'll have to insert this override at the front of the overrides map (I believe), and should use the PTransformMatchers.writeWithRunnerDeterminedSharding() matcher

aaltay · 2017-03-16T17:02:54Z

@reuvenlax any updates here?

davorbonaci · 2017-04-17T16:51:30Z

@reuvenlax, any updates perhaps?

coveralls · 2017-05-10T07:32:51Z

Coverage increased (+0.007%) to 70.491% when pulling 4a605fd on reuvenlax:streaming_auto_shard_write into 8312b6f on apache:master.

coveralls · 2017-05-10T07:33:30Z

Coverage increased (+0.02%) to 70.5% when pulling 4a605fd on reuvenlax:streaming_auto_shard_write into 8312b6f on apache:master.

coveralls · 2017-05-10T07:33:54Z

Coverage increased (+0.07%) to 70.553% when pulling 4a605fd on reuvenlax:streaming_auto_shard_write into 8312b6f on apache:master.

jkff · 2017-05-10T16:01:20Z

There's still that PubsubIO commit. Please rebase -i github/master and delete the commit; if that doesn't work, do something like: git checkout github/master; git cherry-pick $commit (for every useful commit in this PR); git checkout -B streaming_auto_shard_write
There's a lot of formatting changes (the bulk of this PRs diff is formatting changes), roll them back?
Should this apply also to other streaming runners - Spark, Flink, Apex?
I think we're trying to minimize use of PipelineOptions at construction time. Would it make sense to access options.getMaxNumWorkers() at runtime instead, via a do-once ParDo or something?

tgroh · 2017-05-10T20:46:21Z

WRT PipelineOptions at construction time, at the point a PipelineRunner is executing (including in PipelineRunner#run methods), options are available, so that change is fine.

coveralls · 2017-05-11T08:24:51Z

Coverage increased (+0.007%) to 70.195% when pulling 03e9fbf on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

reuvenlax · 2017-05-11T08:37:48Z

PubSubIO commit is dropped
Rolled back unrelated formatting changes.
I don't want to necessarily apply this to other streaming runners, as I'm not 100% whether it's necessary there. I would prefer experts in those runner decide whether this makes sense, and apply this to those runner if necessary.
tgroh already addressed this. We are already inside PipelineRunner here.

coveralls · 2017-05-11T09:46:16Z

Coverage decreased (-0.01%) to 70.174% when pulling 36fceb5 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

coveralls · 2017-05-11T09:50:47Z

Coverage decreased (-7.0e-05%) to 70.188% when pulling 36fceb5 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

coveralls · 2017-05-11T10:05:24Z

Coverage increased (+0.003%) to 70.192% when pulling 36fceb5 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

coveralls · 2017-05-11T10:07:17Z

Coverage decreased (-0.004%) to 70.185% when pulling 36fceb5 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

jkff

Thanks!

jkff · 2017-05-12T00:49:06Z

...e-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

+
+    assertThat(
+        factory.getReplacementTransform(originalApplication).getTransform(),
+        not(equalTo((Object) original)));


Verify that the replacement transform has the expected number of shards?

jkff · 2017-05-12T00:50:06Z

...oogle-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

+        AppliedPTransform<PCollection<T>, PDone, WriteFiles<T>> transform) {
+        return PTransformReplacement.of(
+            PTransformReplacements.getSingletonMainInput(transform),
+            transform.getTransform().withNumShards(options.getMaxNumWorkers() * 2));


Add a comment here explaining why this is a good choice, and what effect it will have? (how many output files we'll actually get, and how many we would have gotten without this override)

coveralls · 2017-05-13T09:25:55Z

Coverage increased (+0.02%) to 70.212% when pulling 92b31cb on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

tgroh · 2017-05-17T00:03:56Z

retest this please

tgroh · 2017-05-17T00:04:34Z

...oogle-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

@@ -1296,6 +1302,46 @@ private StreamingPubsubIOWriteOverrideFactory(DataflowRunner runner) {
  }

  @VisibleForTesting
+  static class StreamingShardedWriteFactory<T>
+      implements PTransformOverrideFactory<PCollection<T>, PDone, WriteFiles<T>> {
+    static final int DEFAULT_NUM_SHARDS = 10;


Comment on the reasoning here.

tgroh · 2017-05-17T00:04:52Z

...e-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java

+    WriteFiles<Object> replacement = (WriteFiles<Object>)
+        factory.getReplacementTransform(originalApplication).getTransform();
+    assertThat(replacement, not(equalTo((Object) original)));
+    assertThat(replacement.getNumShards().get(), equalTo(20));


Can we test the default with unset "(max)NumWorkers"?

tgroh · 2017-05-17T00:05:02Z

Otherwise, LGTM

coveralls · 2017-05-17T01:12:31Z

Coverage increased (+0.5%) to 70.736% when pulling 0c5cd10 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

coveralls · 2017-05-17T01:40:33Z

Coverage increased (+0.6%) to 70.757% when pulling 0c5cd10 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

reuvenlax · 2017-05-17T05:43:13Z

retest this please.

coveralls · 2017-05-17T06:48:12Z

Coverage increased (+0.6%) to 70.753% when pulling 0c5cd10 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

reuvenlax · 2017-05-18T23:21:49Z

retest this please.

reuvenlax · 2017-05-22T21:59:53Z

retest this please

coveralls · 2017-05-22T23:28:30Z

Coverage increased (+0.5%) to 70.656% when pulling 0c5cd10 on reuvenlax:streaming_auto_shard_write into 15bd3a3 on apache:master.

jkff · 2017-05-24T20:06:50Z

The test failure seems legit https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/org.apache.beam$beam-examples-java/11682/testReport/org.apache.beam.examples/WordCountIT/testE2EWordCount_4/

aaltay · 2017-06-08T16:46:43Z

Hey @reuvenlax any updates on this PR?

…taflow runner default to maxNumWorkers * 2 shards.

coveralls · 2017-06-18T07:37:25Z

Coverage decreased (-0.002%) to 70.693% when pulling 272a18a on reuvenlax:streaming_auto_shard_write into 0cabdf6 on apache:master.

coveralls · 2017-06-18T15:40:37Z

Coverage decreased (-0.01%) to 70.684% when pulling 4f6bace on reuvenlax:streaming_auto_shard_write into 0cabdf6 on apache:master.

reuvenlax · 2017-06-18T16:29:35Z

@tgroh can you help figure out this test error? It's complaining that the Create hidden inside of WriteFiles has not bee overridden.

tgroh · 2017-06-19T19:46:39Z

...oogle-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

@@ -337,6 +340,10 @@ public static DataflowRunner fromOptions(PipelineOptions options) {
                  new ReflectiveRootOverrideFactory(StreamingUnboundedRead.class, this)))
          .add(
              PTransformOverride.of(
+                  PTransformMatchers.writeWithRunnerDeterminedSharding(),


For the test to pass, you need to move this override to occur before the Read.Bounded override

reuvenlax · 2017-06-19T23:02:11Z

retest this please.

jkff · 2017-06-20T21:14:55Z

retest this please

coveralls · 2017-06-20T22:48:11Z

Coverage decreased (-0.01%) to 70.684% when pulling 9b9feb1 on reuvenlax:streaming_auto_shard_write into 0cabdf6 on apache:master.

tgroh reviewed Feb 24, 2017

View reviewed changes

tgroh reviewed Feb 27, 2017

View reviewed changes

reuvenlax force-pushed the streaming_auto_shard_write branch from 0038d59 to 689875c Compare April 20, 2017 19:49

reuvenlax changed the title ~~BEAM-1438 Auto shard streaming sinks~~ BEAM-1438 Allow dynamic sharding when doing windowed file-based writes May 4, 2017

reuvenlax changed the title ~~BEAM-1438 Allow dynamic sharding when doing windowed file-based writes~~ BEAM-1438 Auto shard streaming sinks May 4, 2017

reuvenlax closed this May 4, 2017

reuvenlax force-pushed the streaming_auto_shard_write branch from 689875c to ff6bb35 Compare May 4, 2017 12:25

reuvenlax reopened this May 10, 2017

reuvenlax force-pushed the streaming_auto_shard_write branch from 6e50172 to 4a605fd Compare May 10, 2017 05:38

reuvenlax force-pushed the streaming_auto_shard_write branch from 4a605fd to 03e9fbf Compare May 11, 2017 07:12

reuvenlax force-pushed the streaming_auto_shard_write branch from 3e2bbb2 to 36fceb5 Compare May 11, 2017 08:35

jkff requested changes May 12, 2017

View reviewed changes

tgroh reviewed May 17, 2017

View reviewed changes

reuvenlax added 4 commits June 17, 2017 23:18

If a Write operation requests runner-determined sharding, make the Da…

f6c87eb

…taflow runner default to maxNumWorkers * 2 shards.

Address comments.

c9dedcb

Make sure not to set zero as the number of shards.

b3e0cc7

Address comments.

272a18a

reuvenlax force-pushed the streaming_auto_shard_write branch from 0c5cd10 to 272a18a Compare June 18, 2017 06:18

Use WriteFilesTranslation properly.

4f6bace

tgroh reviewed Jun 19, 2017

View reviewed changes

Reorder overrides.

9b9feb1

asfgit closed this in 7b4f5ee Jun 21, 2017

jklukas mentioned this pull request Nov 29, 2018

File Beam PR for allowing unbounded running without numShards mozilla/gcp-ingestion#247

Open

reuvenlax deleted the streaming_auto_shard_write branch December 9, 2018 23:01

BEAM-1438 Auto shard streaming sinks #1952

BEAM-1438 Auto shard streaming sinks #1952

Uh oh!

Conversation

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

Uh oh!

Uh oh!

asfbot commented Feb 8, 2017

Uh oh!

coveralls commented Feb 8, 2017

Uh oh!

asfbot commented Feb 8, 2017

Uh oh!

dhalperi commented Feb 21, 2017

Uh oh!

tgroh left a comment

Choose a reason for hiding this comment

Uh oh!

reuvenlax commented Feb 24, 2017

Uh oh!

tgroh left a comment

Choose a reason for hiding this comment

Uh oh!

aaltay commented Mar 16, 2017

Uh oh!

davorbonaci commented Apr 17, 2017

Uh oh!

coveralls commented May 10, 2017

Uh oh!

coveralls commented May 10, 2017

Uh oh!

coveralls commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

tgroh commented May 10, 2017

Uh oh!

coveralls commented May 11, 2017

Uh oh!

reuvenlax commented May 11, 2017

Uh oh!

coveralls commented May 11, 2017

Uh oh!

coveralls commented May 11, 2017

Uh oh!

coveralls commented May 11, 2017

Uh oh!

coveralls commented May 11, 2017

Uh oh!

jkff left a comment

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!

coveralls commented May 13, 2017

Uh oh!

tgroh commented May 17, 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!

tgroh commented May 17, 2017

Uh oh!

coveralls commented May 17, 2017

Uh oh!

coveralls commented May 17, 2017

Uh oh!

reuvenlax commented May 17, 2017

Uh oh!

reuvenlax commented Feb 8, 2017 •

edited

Loading