[BEAM-2302] Add spilling code to WriteFiles. #3161

reuvenlax · 2017-05-16T04:43:20Z

This is similar to the fix for BEAM-2154.

coveralls · 2017-05-16T05:56:20Z

Coverage decreased (-0.06%) to 70.159% when pulling 675b4fa on reuvenlax:windowed_file_scalability into 529f251 on apache:master.

jkff

Thanks! Didn't look into the tests in too much detail yet.

jkff · 2017-05-16T22:00:48Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java

-                new ViewOverrideFactory())) /* Uses pardos and GBKs */
+    ImmutableList.Builder<PTransformOverride> builder =
+     ImmutableList.<PTransformOverride>builder();
+    if (!options.isUnitTest()) {


I would prefer the other alternative, where the transform itself has an option whether or not a runner should be allowed to set fixed sharding on it (or more like, perhaps introduce a distinction between "no sharding" and "runner-chosen sharding", and use "no sharding" in the test). Having different behavior for test and non-test code is dangerous.

This approach was suggested by @tgroh, who felt that the transform should never be overridden in unit tests.

Specifically, tests executions for the sdk and runners-core modules (e.g. the validates-runner execution) don't need to modify any behavior of the original transforms, and I'd like to avoid having the transform specify how it is permitted to be overridden.

I would like to pull any test-related options out of DirectOptions and mark them as @Internal and @Hidden, however.

@tgroh where should the options go?

Either in a new DirectTestOptions or in TestPipelineOptions

If you add DirectTestOptions I believe you will have to register them in the DirectPipelineRegistrar as well

jkff · 2017-05-16T23:07:15Z

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

   * for each shard have been collected into a single iterable.
   */
  private class WriteShardedBundles extends DoFn<KV<Integer, Iterable<T>>, FileResult> {
+    boolean setWindowedShardNumber;


Rename the variable to something more clear?

made an enum

jkff · 2017-05-16T23:07:32Z

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

      Writer<T> writer = writeOperation.createWriter();
      if (windowedWrites) {
-        writer.openWindowed(UUID.randomUUID().toString(), window, c.pane(), c.element().getKey());
+        int shardNumber = setWindowedShardNumber ? c.element().getKey() : UNKNOWN_SHARDNUM;


And add a comment here

It's more clear with an enum now, and a detailed comment was added elsewhere

jkff · 2017-05-16T23:08:33Z

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

-              "WriteBundles",
-              ParDo.of(windowedWrites ? new WriteWindowedBundles() : new WriteUnwindowedBundles()));
+      if (windowedWrites) {
+        TupleTag<FileResult> writtenRecordsTag = new TupleTag<FileResult>() {};


Note: these anonymous TupleTag's capture the enclosing class. Might want to use the TupleTag(String) constructor instead.

jkff · 2017-05-16T23:09:37Z

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

+            writeTuple
+                .get(unwrittedRecordsTag)
+                .apply("GroupUnwritten", GroupByKey.<Integer, T>create())
+                .apply("WriteUnwritten", ParDo.of(new WriteShardedBundles(false)))


Instead of a boolean, maybe better to use an enum with eloquently named values

jkff · 2017-05-16T23:10:09Z

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

+  private SimpleSink makeSimpleSink(boolean windowedWrites) {
+    FilenamePolicy filenamePolicy = new PerWindowFiles("file", "simple");
+    return new SimpleSink(getBaseOutputDirectory(), filenamePolicy);
+ //   return new SimpleSink(getBaseOutputDirectory(), "file", "-SS-of-NN", "simple");


Remove commented-out code

jkff · 2017-05-16T23:10:21Z

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

  public void testEmptyWrite() throws IOException {
-    runWrite(Collections.<String>emptyList(), IDENTITY_MAP, getBaseOutputFilename());
+    runWrite(Collections.<String>emptyList(), IDENTITY_MAP, getBaseOutputFilename(),
+        Optional.<Integer>absent(), false);


false /* windowedWrites */

I forget not everyone uses IntelliJ :)

jkff · 2017-05-16T23:10:37Z

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

  }
-  private SimpleSink makeSimpleSink() {
-    return new SimpleSink(getBaseOutputDirectory(), "file", "-SS-of-NN", "simple");
+  private SimpleSink makeSimpleSink(boolean windowedWrites) {


This is not using windowedWrites.

jkff · 2017-05-16T23:14:10Z

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

    }

-    SimpleSink sink = makeSimpleSink();
+    SimpleSink sink = makeSimpleSink(windowedWrites);


I think it will be easier to read all the tests if, instead of passing the parameters such as numConfiguredShards, windowedWrites etc., you make runShardedWrite take a WriteFiles instance directly.

It's easier to read WriteFiles.to(makeSimpleSink(true) /* windowedWrites */).withMaxNumWritersPerBundle(42) than "..., true, Optional.of(42)".

coveralls · 2017-05-17T02:14:47Z

Coverage increased (+0.5%) to 70.679% when pulling a5cdacf on reuvenlax:windowed_file_scalability into 529f251 on apache:master.

reuvenlax · 2017-05-18T23:30:08Z

retest this please.

coveralls · 2017-05-19T00:53:06Z

Coverage increased (+0.5%) to 70.712% when pulling a5cdacf on reuvenlax:windowed_file_scalability into 529f251 on apache:master.

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

retest this please.

coveralls · 2017-05-22T23:26:35Z

Coverage increased (+0.4%) to 70.585% when pulling a5cdacf on reuvenlax:windowed_file_scalability into 529f251 on apache:master.

coveralls · 2017-05-24T17:45:08Z

Coverage increased (+0.4%) to 70.628% when pulling 597f8a8 on reuvenlax:windowed_file_scalability into 529f251 on apache:master.

jkff

Thanks!

jkff · 2017-05-24T18:36:18Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java

-                PTransformMatchers.classEqualTo(CreatePCollectionView.class),
-                new ViewOverrideFactory())) /* Uses pardos and GBKs */
+    TestPipelineOptions testOptions = options.as(TestPipelineOptions.class);
+    ImmutableList.Builder<PTransformOverride> builder =


jkff · 2017-05-24T18:38:47Z

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

   */
  public WriteFiles<T> withNumShards(ValueProvider<Integer> numShardsProvider) {
-    return new WriteFiles<>(sink, null, numShardsProvider, windowedWrites);
+    return new WriteFiles<>(sink, null, numShardsProvider, windowedWrites,


might be time to convert this class to AutoValue (but feel free to ignore)

I think best left for another PR

jkff · 2017-05-24T18:42:16Z

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

+          windowedWriters.put(key, writer);
+          LOG.debug("Done opening writer");
+        } else {
+          c.output(unwrittedRecordsTag, KV.of(


nitpick: might make sense to put this (attaching a sharding key) into its own DoFn, and here output simply to a TupleTag<T>. That DoFn can be implemented more efficiently, without generating a random number for every value (you can generate a single number in startBundle or even setup(), and then keep incrementing it for every element - this has the same statistical properties).

Also feel free to ignore.

Think making it a separate DoFn is overkill, but the suggestion of caching the random number is a good one (and what we do elsewhere)

jkff · 2017-05-24T18:49:38Z

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

-              ParDo.of(windowedWrites ? new WriteWindowedBundles() : new WriteUnwindowedBundles()));
+      if (windowedWrites) {
+        TupleTag<FileResult> writtenRecordsTag =
+            new TupleTag<FileResult>("writtenRecordsTag") {};


These anonymous TupleTag classes capture the enclosing transform so they'll blow up the serialized graph unnecessarily. I think the code will work even if you remove {} - if it won't, you'll probably just need to specify some coders, because all the {} does is capture the type information.

jkff · 2017-05-24T18:52:15Z

sdks/java/core/src/main/java/org/apache/beam/sdk/testing/TestPipelineOptions.java

+  @Internal
+  @Hidden
+  @org.apache.beam.sdk.options.Description(
+      "Indicates whether this is an automatically-run unit test.")


...as opposed to? (what else is TestPipelineOptions used for, if not unit tests?)

"automatically run"

jkff · 2017-05-24T18:55:53Z

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

        IDENTITY_MAP,
        getBaseOutputFilename(),
-        Optional.of(20));
+        Optional.of(20),


I'm still not super happy about readability here due to how many parameters there are, and how easy they are to confuse. What do you think about my suggestion to pass WriteFiles directly, so that the call site contains a builder call and it's obvious what parameters are configured with what values?

aaltay · 2017-06-08T16:47:08Z

Hey @reuvenlax any updates on this PR?

reuvenlax · 2017-06-08T20:34:47Z

@jkff comments are addressed.

coveralls · 2017-06-08T21:45:48Z

Coverage decreased (-0.09%) to 70.502% when pulling c854d34 on reuvenlax:windowed_file_scalability into 4aef938 on apache:master.

coveralls · 2017-06-08T22:05:08Z

Coverage decreased (-0.08%) to 70.509% when pulling c854d34 on reuvenlax:windowed_file_scalability into 4aef938 on apache:master.

jkff · 2017-06-13T21:34:02Z

Run Dataflow ValidatesRunner

reuvenlax · 2017-06-16T21:14:30Z

The dataflow test failure seems unrelated to this change.

reuvenlax · 2017-06-16T21:14:36Z

Run Dataflow ValidatesRunner

reuvenlax · 2017-06-17T02:12:06Z

Again fails in CombineTest.testSimpleCombineWithContext on Dataflow Runner. I see no way this could be related.

reuvenlax · 2017-06-19T20:56:39Z

Run Dataflow ValidatesRunner

jkff requested changes May 16, 2017

View reviewed changes

jkff requested changes May 24, 2017

View reviewed changes

reuvenlax added 5 commits June 8, 2017 13:05

Add spilling code to WriteFiles.

cb9973e

Address comments.

1ae3ab7

Fix output tags.

68b9eb1

Move flag to TestPipelineOptions.

b625e34

Address comments.

bce2c12

reuvenlax force-pushed the windowed_file_scalability branch from 597f8a8 to bce2c12 Compare June 8, 2017 20:18

Refactor test per comments.

c854d34

asfgit closed this in 698b89e Jun 20, 2017

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

[BEAM-2302] Add spilling code to WriteFiles. #3161

[BEAM-2302] Add spilling code to WriteFiles. #3161

Uh oh!

Conversation

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

Uh oh!

Uh oh!

coveralls commented May 16, 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!

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!

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!

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 17, 2017

Uh oh!

reuvenlax commented May 18, 2017

Uh oh!

coveralls commented May 19, 2017

Uh oh!

reuvenlax commented May 22, 2017

Uh oh!

coveralls commented May 22, 2017

Uh oh!

coveralls commented May 24, 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!

Choose a reason for hiding this comment

reuvenlax commented May 16, 2017 •

edited

Loading