[BEAM-1432] Inject number of Shards as a Side Input to Write #1941

tgroh · 2017-02-07T23:20:17Z

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.

This permits users to pass a PTransform from the input elements to the
number of shards instead of requiring a constant amount. This enables
sharding to be determined based on the input data rather than a constant
value.

This can also convert the DirectRunner to inject a custom sharding
strategy in its override, rather than relying on bundling behavior.

This permits users to pass a PTransform from the input elements to the number of shards instead of requiring a constant amount. This enables sharding to be determined based on the input data rather than a constant value. Tests still should be written. This can also convert the DirectRunner to inject a custom sharding strategy in its override, rather than relying on custom bundling.

tgroh · 2017-02-07T23:20:38Z

Tests still need to be written. @dhalperi @reuvenlax you two are likely interested.

asfbot · 2017-02-07T23:24:02Z

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

dhalperi

It might be easy to extend to support ValueProvider for numShards.

dhalperi · 2017-02-07T23:42:24Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

-      return numShards;
+          .addIfNotNull(
+              DisplayData.item("sharding", getSharding() == null ? null : getSharding().getClass()))
+          .include("sharding", getSharding());


indent seems off, add a label.

Doesn't .include do L122 already?

Cleaned this up. include does indeed do that, but also doesn't permit null (and getSharding can of course be null)

dhalperi · 2017-02-07T23:42:30Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

      this.sink = sink;
-      this.numShards = numShards;
+        this.computeNumShards = computeNumShards;


indent off?

dhalperi · 2017-02-07T23:42:44Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

@@ -96,11 +95,11 @@
   */
  public static class Bound<T> extends PTransform<PCollection<T>, PDone> {
    private final Sink<T> sink;
-    private int numShards;
+    private final PTransform<PCollection<T>, PCollectionView<Integer>> computeNumShards;


@Nullable

dhalperi · 2017-02-07T23:43:01Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

@@ -140,6 +130,10 @@ public int getNumShards() {
      return sink;
    }

+    public PTransform<PCollection<T>, PCollectionView<Integer>> getSharding() {


@Nullable, javadoc

dhalperi · 2017-02-07T23:49:19Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

+
+    @Override
+    public void populateDisplayData(DisplayData.Builder builder) {
+      builder.add(DisplayData.item("numShards", numShards));


dhalperi · 2017-02-07T23:49:54Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/WriteTest.java

@@ -254,14 +254,11 @@ public void testWriteWithSessions() {
  public void testBuildWrite() {
    Sink<String> sink = new TestSink() {};
    Write.Bound<String> write = Write.to(sink).withNumShards(3);
-    assertEquals(3, write.getNumShards());


reuvenlax · 2017-02-08T01:27:30Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

+     * Returns a new {@link Write.Bound} that will write to the current {@link Sink} with
+     * runner-determined sharding.
+     */
+    public Bound<T> withoutSharding() {


withDynamicSharding?

withRunnerDeterminedSharding is more precise (as DynamicSharding can be user-determined)

coveralls · 2017-02-08T02:50:03Z

Coverage decreased (-0.007%) to 69.598% when pulling a33c457 on tgroh:write_inject_sharding into 8c1a577 on apache:master.

asfbot · 2017-02-08T02:50:38Z

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

asfbot · 2017-02-09T17:50:20Z

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

asfbot · 2017-02-09T18:53:51Z

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

coveralls · 2017-02-09T19:00:29Z

Coverage decreased (-0.02%) to 69.697% when pulling 41071dd on tgroh:write_inject_sharding into 2eeeaa2 on apache:master.

tgroh · 2017-02-10T00:42:10Z

R: @dhalperi

I have also added the ValueProvider alternative

dhalperi

essentially LGTM, minor comments.

If you're happy with the validation improvements, go ahead and self-merge. If you want me to TAL, let me know.

dhalperi · 2017-02-17T08:13:05Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

@@ -151,7 +160,42 @@ public int getNumShards() {
     * runner-controlled sharding.


consistent language? (runner-determined/controlled)

I like determined. Done.

dhalperi · 2017-02-17T08:16:06Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

+              if (numShards == null) {
+                minShardsNeeded = 1;
+              } else {
+                minShardsNeeded = c.sideInput(numShards);


validate this value.

dhalperi · 2017-02-17T08:17:54Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

+
+    @Override
+    public void populateDisplayData(DisplayData.Builder builder) {
+      builder.add(DisplayData.item("numShards", numShards).withLabel("ConstantShards"));


"Fixed Number of Shards" to match the existing label?

dhalperi · 2017-02-17T08:19:00Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/Write.java

-      public Integer apply(T input) {
+      @ProcessElement
+      public void processElement(ProcessContext context) {
+        Integer shardCount = context.sideInput(numShards);


validate this number.

coveralls · 2017-02-17T19:16:06Z

Coverage decreased (-0.5%) to 69.192% when pulling 2a642f5 on tgroh:write_inject_sharding into 2eeeaa2 on apache:master.

asfbot · 2017-02-17T19:17:17Z

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

Failed Tests: 0

--none--

dhalperi · 2017-02-17T22:40:05Z

(please self-merge at will)

coveralls · 2017-02-17T23:42:49Z

Coverage decreased (-0.5%) to 69.189% when pulling 321b095 on tgroh:write_inject_sharding into 2eeeaa2 on apache:master.

asfbot · 2017-02-17T23:43:27Z

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

dhalperi reviewed Feb 7, 2017

View reviewed changes

reuvenlax reviewed Feb 8, 2017

View reviewed changes

tgroh added 2 commits February 7, 2017 18:07

fixup! Inject number of Shards as a Side Input to Write

81079b9

fixup! Inject number of Shards as a Side Input to Write

a33c457

tgroh changed the title ~~Inject number of Shards as a Side Input to Write~~ [BEAM-1432] Inject number of Shards as a Side Input to Write Feb 8, 2017

fixup! Inject number of Shards as a Side Input to Write

41071dd

tgroh force-pushed the write_inject_sharding branch from cdc1840 to 41071dd Compare February 9, 2017 17:51

dhalperi approved these changes Feb 17, 2017

View reviewed changes

fixup! Inject number of Shards as a Side Input to Write

2a642f5

fixup! Inject number of Shards as a Side Input to Write

321b095

asfgit closed this in 918c7fa Feb 18, 2017

		@@ -151,7 +160,42 @@ public int getNumShards() {
		* runner-controlled sharding.

[BEAM-1432] Inject number of Shards as a Side Input to Write #1941

[BEAM-1432] Inject number of Shards as a Side Input to Write #1941

Conversation

tgroh commented Feb 7, 2017 • edited

tgroh commented Feb 7, 2017

asfbot commented Feb 7, 2017

dhalperi 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

coveralls commented Feb 8, 2017

asfbot commented Feb 8, 2017

asfbot commented Feb 9, 2017

asfbot commented Feb 9, 2017

coveralls commented Feb 9, 2017

tgroh commented Feb 10, 2017

dhalperi 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

coveralls commented Feb 17, 2017

asfbot commented Feb 17, 2017

Failed Tests: 0

dhalperi commented Feb 17, 2017

coveralls commented Feb 17, 2017

asfbot commented Feb 17, 2017

tgroh commented Feb 7, 2017 •

edited