[BEAM-2052] Allow dynamic sharding in windowed file sinks by jkff · Pull Request #3023 · apache/beam

jkff · 2017-05-09T23:40:34Z

This is a slightly modified and rearranged version of @reuvenlax 's #2647 .

My concerns about it are:

In direct runner, the integration tests of dynamic sharding are vacuous, because direct runner replaces unspecified sharding with fixed sharding at https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java (applied at https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java#L217). However, this is a testing-only concern: other runners don't have this override, so overall the testing is non-vacuous, this is just hard to test against direct runner and I suspect that we probably want these tests to be non-vacuous in direct runner too.
When I removed that override for testing purposes, I noticed that there's a very large number of files being written - primarily, I guess, because the bundles are very small. So large a number of files that the test time for batch with dynamic sharding grows from 21 seconds to 5 minutes. In particular, we write many, many files for each window/pane - presumably because in streaming runners and in direct runner, there's at least 1 bundle per key, and we create at least 1 file per bundle in WriteFiles.Write(Windowed,Unwindowed)Bundles.

Reuven, can you please comment on whether this "at least 1 file per key" is expected behavior in a streaming runner? I suspect that it's not, but then I'm not sure how to fix the PR semantically.

CC: @reuvenlax @davorbonaci @dhalperi

FileBasedWriteOperation -> WriteOperation, FileBasedWriter -> Writer

jkff · 2017-05-10T04:24:14Z

retest this please

reuvenlax · 2017-05-10T06:18:47Z

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

+
+    @FinishBundle
+    public void finishBundle(FinishBundleContext c) throws Exception {
+      FileResult result = writer.close();


if (writer != null) {
}

(empty bundles are legal I believe)

reuvenlax · 2017-05-10T06:31:40Z

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

-    private FileBasedWriter<T> writer = null;
-    private BoundedWindow window = null;
+  private class WriteWindowedBundles extends DoFn<T, FileResult> {
+    private Map<KV<BoundedWindow, PaneInfo>, Writer<T>> windowedWriters;


incorrect I think - BoundedWindow isn't guaranteed to implement a proper hashCode. That's why I used Coder.structuralValue in the original PR

In particular while this will work for our built-in windows, users can write their own window functions, and we do not insist that they provide hashCode - all we insist is that they provide a Coder. Hence the reason I used Coder.structuredValue to encode the windows.

BoundedWindow documents that it must implement equals and hashCode (https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/BoundedWindow.java#L30), and there's a lot of HashMaps keyed with BoundedWindow in the repo.

reuvenlax · 2017-05-10T12:29:45Z

1. This is true for the non-windowed case as well. Fixing this might require some work in the DirectRunner - @tgroh can comment. 2. 1 bundle per key is true of the Dataflow streaming runner, but is in general runner dependent. This is true of unwindowed FileBasedSinks as well, this is not new in the windowing support. I see two regressions in your PR from mine (one of which was also caught by a test). Added comments.

…

On Tue, May 9, 2017 at 4:40 PM, Eugene Kirpichov ***@***.***> wrote: This is a slightly modified and rearranged version of @reuvenlax <https://github.com/reuvenlax> 's #2647 <#2647> . My concerns about it are: 1. In direct runner, the integration tests of dynamic sharding are vacuous, because direct runner replaces unspecified sharding with fixed sharding at https://github.com/apache/beam/blob/master/runners/ direct-java/src/main/java/org/apache/beam/runners/direct/ WriteWithShardingFactory.java <https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/WriteWithShardingFactory.java> (applied at https://github.com/apache/beam/blob/master/runners/ direct-java/src/main/java/org/apache/beam/runners/direct/ DirectRunner.java#L217 <https://github.com/apache/beam/blob/master/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectRunner.java#L217>). However, this is a testing-only concern: other runners don't have this override, so overall the testing is non-vacuous, this is just hard to test against direct runner and I suspect that we probably want these tests to be non-vacuous in direct runner too. 2. When I removed that override for testing purposes, I noticed that there's a very large number of files being written - primarily, I guess, because the bundles are very small. So large a number of files that the test time for batch with dynamic sharding grows from 21 seconds to 5 minutes. In particular, we write many, many files for each window/pane - presumably because in streaming runners and in direct runner, there's at least 1 bundle per key, and we create at least 1 file per bundle in WriteFiles.Write(Windowed,Unwindowed)Bundles. Reuven, can you please comment on whether this "at least 1 file per key" is expected behavior in a streaming runner? I suspect that it's not, but then I'm not sure how to fix the PR semantically. CC: @reuvenlax <https://github.com/reuvenlax> @davorbonaci <https://github.com/davorbonaci> @dhalperi <https://github.com/dhalperi> ------------------------------ You can view, comment on, or merge this pull request online at: #3023 Commit Summary - Implement dynamic-sharding for windowed file outputs, and add an integration test. - Renames FileBasedSink inner classes - Simpler code for setting shard numbers on results in FileBasedSink - Splits WriteBundles into windowed/unwindowed versions File Changes - *M* examples/java/src/main/java/org/apache/beam/examples/ WindowedWordCount.java <https://github.com/apache/beam/pull/3023/files#diff-0> (7) - *M* examples/java/src/main/java/org/apache/beam/examples/ common/WriteOneFilePerWindow.java <https://github.com/apache/beam/pull/3023/files#diff-1> (24) - *M* examples/java/src/test/java/org/apache/beam/examples/ WindowedWordCountIT.java <https://github.com/apache/beam/pull/3023/files#diff-2> (29) - *M* runners/core-construction-java/src/test/java/org/apache/ beam/runners/core/construction/PTransformMatchersTest.java <https://github.com/apache/beam/pull/3023/files#diff-3> (2) - *M* runners/direct-java/src/test/java/org/apache/beam/runners/direct/ WriteWithShardingFactoryTest.java <https://github.com/apache/beam/pull/3023/files#diff-4> (2) - *M* runners/google-cloud-dataflow-java/src/main/java/org/apache/ beam/runners/dataflow/util/DefaultCoderCloudObjectTransla torRegistrar.java <https://github.com/apache/beam/pull/3023/files#diff-5> (2) - *M* sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSink.java <https://github.com/apache/beam/pull/3023/files#diff-6> (14) - *M* sdks/java/core/src/main/java/org/apache/beam/sdk/io/ FileBasedSink.java <https://github.com/apache/beam/pull/3023/files#diff-7> (316) - *M* sdks/java/core/src/main/java/org/apache/beam/sdk/io/ TFRecordIO.java <https://github.com/apache/beam/pull/3023/files#diff-8> (16) - *M* sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSink.java <https://github.com/apache/beam/pull/3023/files#diff-9> (14) - *M* sdks/java/core/src/main/java/org/apache/beam/sdk/io/ WriteFiles.java <https://github.com/apache/beam/pull/3023/files#diff-10> (259) - *M* sdks/java/core/src/test/java/org/apache/beam/sdk/io/ FileBasedSinkTest.java <https://github.com/apache/beam/pull/3023/files#diff-11> (162) - *M* sdks/java/core/src/test/java/org/apache/beam/sdk/io/ SimpleSink.java <https://github.com/apache/beam/pull/3023/files#diff-12> (4) - *M* sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/ xml/XmlSink.java <https://github.com/apache/beam/pull/3023/files#diff-13> (8) Patch Links: - https://github.com/apache/beam/pull/3023.patch - https://github.com/apache/beam/pull/3023.diff — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3023>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AUGE1ef1ECr_2_E3BIuSK-oH13NV2a1Lks5r4Pl5gaJpZM4NWAz-> .

reuvenlax · 2017-05-10T15:35:13Z

examples/java/src/test/java/org/apache/beam/examples/WindowedWordCountIT.java

+    testWindowedWordCountPipeline(options);
+  }
+
+  @Test


I think we should remove this test until we fix the streaming runner to not generate a file per bundle. Until then this is not a suggested use case for streaming, so I think it's ok to remove the test.

…gration test.

jkff · 2017-05-10T17:50:52Z

Run Spark ValidatesRunner

jkff · 2017-05-10T17:51:06Z

Run Flink ValidatesRunner

jkff · 2017-05-10T17:51:14Z

Run Dataflow ValidatesRunner

jkff · 2017-05-10T19:23:12Z

I merged this after manually running WindowedWordCountIT with Flink runner.

Looking at the dependency error - it looks pretty clearly unrelated to my PR though, the diff doesn't even mention "findbugs"...

jkff · 2017-05-10T19:24:12Z

Sorry, I'm an idiot, I actually introduced the error. Will send a fix PR right away.

jkff · 2017-05-10T19:29:50Z

DO NOT cherrypick without #3059

Renames FileBasedSink inner classes

c43cc4a

FileBasedWriteOperation -> WriteOperation, FileBasedWriter -> Writer

reuvenlax reviewed May 10, 2017

View reviewed changes

reuvenlax and others added 3 commits May 10, 2017 10:48

Implement dynamic-sharding for windowed file outputs, and add an inte…

bbaadb6

…gration test.

Simpler code for setting shard numbers on results in FileBasedSink

fedeb10

Splits WriteBundles into windowed/unwindowed versions

b91e36b

jkff force-pushed the finish-pr-2647-2 branch from b775df1 to b91e36b Compare May 10, 2017 17:48

asfgit closed this in d0085e6 May 10, 2017

jkff deleted the finish-pr-2647-2 branch May 10, 2017 19:27

jkff mentioned this pull request May 10, 2017

Adds dependency on findbugs to examples/java #3059

Merged

jkff added a commit to jkff/incubator-beam that referenced this pull request May 10, 2017

This closes apache#3023

213eef8

jkff mentioned this pull request May 10, 2017

[BEAM-2052] Cherrypick #3023 and #3059 to release-2.0.0 #3060

Closed

asfgit pushed a commit that referenced this pull request May 10, 2017

This closes #3023

9311e18

jkff mentioned this pull request May 10, 2017

[BEAM-2052] Allow dynamic sharding in windowed file sinks #2647

Closed

Conversation

jkff commented May 9, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

reuvenlax May 10, 2017

Choose a reason for hiding this comment

Uh oh!

reuvenlax May 10, 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!

reuvenlax May 10, 2017

Choose a reason for hiding this comment

Uh oh!

jkff May 10, 2017

Choose a reason for hiding this comment

Uh oh!

reuvenlax commented May 10, 2017 via email

Uh oh!

reuvenlax May 10, 2017

Choose a reason for hiding this comment

Uh oh!

jkff commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

jkff commented May 10, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

reuvenlax May 10, 2017 •

edited

Loading