Many simplifications to WriteFiles #4145

jkff · 2017-11-17T20:42:27Z

Commits explain what's going on. I recommend reviewing each commit individually. Most importantly, this unifies windowed and unwindowed finalize (this is the only meaningful change - everything else is just restructuring), and refactors the transform into sub-transforms for better readability.

Many more simplifications are possible in WriteOperation/FileBasedSink themselves, but I'll defer that to post-FileIO #3817 (this PR can be reviewed in parallel with that one)

R: @reuvenlax or feel free to reassign to anybody else.

asfgit · 2017-11-17T21:33:35Z

SUCCESS

--none--

asfgit · 2017-11-17T21:51:02Z

FAILURE

--none--

asfgit · 2017-11-17T22:07:27Z

SUCCESS

--none--

jkff · 2017-11-20T21:48:01Z

@chamikaramj agreed to take a look.

reuvenlax · 2017-11-21T00:37:13Z

FYI, I've been going through this PR as well.

…

On Tue, Nov 21, 2017 at 5:48 AM, Eugene Kirpichov ***@***.***> wrote: @chamikaramj <https://github.com/chamikaramj> agreed to take a look. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4145 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AUGE1QI94jMDbDJ5TwaNgeFdSWSdSvn1ks5s4fOagaJpZM4QimNq> .

reuvenlax · 2017-11-22T18:15:50Z

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

+                ? new FileResult<>(
+                    writer.getOutputFile(), UNKNOWN_SHARDNUM, window, key.paneInfo, key.destination)
+                : new FileResult<>(
+                    writer.getOutputFile(), UNKNOWN_SHARDNUM, null, null, key.destination);


Don't we lose the shard number, where before we had the shard number in the output?

Not sure what you mean here: this is the unsharded WriteBundles, which emits normally-written bundles with UNKNOWN_SHARDNUM and emits spilled data with a shard number (that is later discarded). This was the case before this change, and is still the case after this change.

reuvenlax · 2017-11-22T18:17:58Z

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

+            shardNumberAssignment == ShardAssignment.ASSIGN_WHEN_WRITING
+                ? c.element().getKey().getShardNumber()
+                : UNKNOWN_SHARDNUM;
+        if (windowedWrites) {


BTW, we simplified the FileBasedSink API at the expense of extra complexity in multiple places in WriteFiles. Is this really a simplification?

This is an intermediate step to more simplifications that will come after FileIO - in this case, I removed result from Writer in order to simplify removing DestinationT from Writer (and later also removing UserT), to end up with having Writer be a bare-bones consumer for format-specific records that are directly written into it, e.g. for text files it is strings, without any user or destination types involved.

I think this is also desirable from a readability point of view: shard number, destination etc. are bookkeeping information private to WriteFiles, and it's better if they are managed by WriteFiles rather than scattered across two classes.

reuvenlax · 2017-11-22T18:19:31Z

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

@@ -854,7 +854,8 @@ public void processElement(ProcessContext c) {
        } else if (numShardsProvider != null) {
          fixedNumShards = numShardsProvider.get();
        } else {
-          fixedNumShards = null;


Enforcing something like this at runtime is unfortunate. Any way to enforce at graph construction time?

reuvenlax · 2017-11-22T18:34:04Z

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

-      TupleTag<KV<ShardedKey<Integer>, UserT>> unwrittedRecordsTag =
-          new TupleTag<>("unwrittenRecordsTag");
+      TupleTag<KV<ShardedKey<Integer>, UserT>> spilledRecordsTag =
+          new TupleTag<>("spilledRecordsTag");


changing the TupleTag name breaks update compatibility. Update compatibility is the whole reason we used an explicit name here in the first place.

Let's add comments to here and other places so that update compatibility is not accidentally broken by future updates.

reuvenlax · 2017-11-22T18:34:59Z

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

+              // number assigned at all. Drop the shard number on the spilled records so that
+              // shard numbers are assigned together to both the spilled and non-spilled files in
+              // finalize.
+              .apply("GroupSpilled", GroupByKey.<ShardedKey<Integer>, UserT>create())


again, we can't change the transform name, it is meant to be stable.

reuvenlax · 2017-11-22T18:39:22Z

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

-              .setCoder(FileResultCoder.of(shardedWindowCoder, destinationCoder));
+                  "WriteSpilled", ParDo.of(new WriteShardedBundles()).withSideInputs(sideInputs))
+              .setCoder(fileResultCoder)
+              .apply("DropShardNum", ParDo.of(


why drop shard numbers here?

reuvenlax · 2017-11-22T18:53:22Z

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

-                  ParDo.of(new WriteShardedBundles(ShardAssignment.ASSIGN_IN_FINALIZE))
-                      .withSideInputs(sideInputs))
-              .setCoder(FileResultCoder.of(shardedWindowCoder, destinationCoder));
+                  "WriteSpilled", ParDo.of(new WriteShardedBundles()).withSideInputs(sideInputs))


so you're now going to assign shard #s based on the key? This is potentially broken for batch, as some of the files (the ones written bye WriteWindowedBundles) are assigned in finalize, but the spilled files are assigned here. I worry that could lead to collisions.

jkff

Github isn't letting me respond inline:

I reverted the rename of tuple tags (spilled -> unwritten).
Regarding shard number assignment for spilled records: in the old code, it was already the case that written records have UNKNOWN_SHARDNUM and unwritten (spilled) have a shard number assigned according to spill factor, and then in finalize with ASSIGN_IN_FINALIZE both of these shard numbers are actually discarded and recomputed. I wanted the finalize operation to not receive a mix of known and unknown shard numbers, but we still need to shard the spilled records - so I moved the logic of discarding shard numbers into an explicit DoFn DropShardNum.

For this comment:

Enforcing something like this at runtime is unfortunate. Any way to enforce at graph construction time?
Hide outdated

the original context is lost. Is it still relevant?

jkff · 2017-11-23T00:25:54Z

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

+                ? new FileResult<>(
+                    writer.getOutputFile(), UNKNOWN_SHARDNUM, window, key.paneInfo, key.destination)
+                : new FileResult<>(
+                    writer.getOutputFile(), UNKNOWN_SHARDNUM, null, null, key.destination);


Not sure what you mean here: this is the unsharded WriteBundles, which emits normally-written bundles with UNKNOWN_SHARDNUM and emits spilled data with a shard number (that is later discarded). This was the case before this change, and is still the case after this change.

jkff · 2017-11-23T00:30:08Z

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

+            shardNumberAssignment == ShardAssignment.ASSIGN_WHEN_WRITING
+                ? c.element().getKey().getShardNumber()
+                : UNKNOWN_SHARDNUM;
+        if (windowedWrites) {


This is an intermediate step to more simplifications that will come after FileIO - in this case, I removed result from Writer in order to simplify removing DestinationT from Writer (and later also removing UserT), to end up with having Writer be a bare-bones consumer for format-specific records that are directly written into it, e.g. for text files it is strings, without any user or destination types involved.

I think this is also desirable from a readability point of view: shard number, destination etc. are bookkeeping information private to WriteFiles, and it's better if they are managed by WriteFiles rather than scattered across two classes.

chamikaramj

Thanks.

chamikaramj · 2017-11-21T03:09:45Z

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

@@ -868,53 +934,10 @@ protected void finishWrite() throws Exception {}
     * id populated for the case of static sharding. In cases where the runner is dynamically


Top part of this comment ("Performs bundle initialization. For example..") seems to be more generic than it should be. This is probably from Sink.java days. Shall we simplify/update this (and possibly other) doc comments here ?

chamikaramj · 2017-11-23T00:52:50Z

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

+                  writer.getOutputFile(),
+                  shard,
+                  GlobalWindow.INSTANCE,
+                  PaneInfo.ON_TIME_AND_ONLY_FIRING,


Are we changing the default values chosen for PainInfo here (could not easily find out by following the code) ?

Previously the code was structured differently, and the values passed in this particular codepath ended up being ignored. I consolidated things somewhat to handle much of windowed and unwindowed case the same way, and made the requirements more strict, in particular that window and pane have to be always set.

chamikaramj · 2017-11-23T01:45:19Z

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

-        c.output(
-            new FileResult<>(
-                writer.getOutputFile(), shardNumber, window, c.pane(), entry.getKey()));
+        int shard = c.element().getKey().getShardNumber();


chamikaramj · 2017-11-23T01:51:07Z

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

-    final PCollectionView<Integer> numShardsView;
+    PCollectionView<Integer> numShardsView =
+        (computeNumShards == null) ? null : input.apply(computeNumShards);
+    List<PCollectionView<Integer>> shardingSideInputs = numShardsView == null


(numShardsView == null) for readability.

chamikaramj · 2017-11-23T01:53:44Z

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

-      TupleTag<KV<ShardedKey<Integer>, UserT>> unwrittedRecordsTag =
-          new TupleTag<>("unwrittenRecordsTag");
+      TupleTag<KV<ShardedKey<Integer>, UserT>> spilledRecordsTag =
+          new TupleTag<>("spilledRecordsTag");


Let's add comments to here and other places so that update compatibility is not accidentally broken by future updates.

chamikaramj · 2017-11-23T02:07:10Z

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

-        LOG.info("No output files to write.");
+      LOG.debug("Copying {} files.", numFiles);
+      List<ResourceId> srcFiles = new ArrayList<>(resultsToFinalFilenames.size());
+      List<ResourceId> dstFiles = new ArrayList<>(resultsToFinalFilenames.size());


Assert that list sizes are equal.

Seems overkill - they are created right next to each other in code, and FileSystems.copy() already does that verification. I removed the size hints to make it a little simpler (preallocation probably doesn't matter here).

reuvenlax · 2017-11-22T18:58:32Z

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

-              "When finalizing a windowed write, should have set fixed sharding");
-        }
+        fixedNumShards = getFixedNumShards.apply(c);
+        checkState(fixedNumShards != null, "Windowed write should have set fixed sharding");


Windowed (non triggered) writes in batch do not need fixed sharding

After #4124 they do - see #4137 for explanation.

reuvenlax · 2017-11-25T00:15:52Z

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

+    PCollection<FileResult<DestinationT>> tempFileResults =
+        (computeNumShards == null && numShardsProvider == null)
+            ? input.apply(
+                "WriteUnshardedBundlesToTempFiles",


Unfortunately, refactoring into new PTransforms changes the name of every single sub step (since step names are hierarchical).

jkff

Thanks, addressed all comments; PTAL.

I replied inline where I could. To address Reuven's comment about refactoring into new PTransforms: per offline discussion, this can be handled by publishing instructions for applying a transform mapping in release notes; I would not like this issue to block simplifications of this code, whose complexity has unfortunately already bitten us many times in more painful ways than update incompatibility.

jkff · 2017-11-27T19:04:58Z

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

-        LOG.info("No output files to write.");
+      LOG.debug("Copying {} files.", numFiles);
+      List<ResourceId> srcFiles = new ArrayList<>(resultsToFinalFilenames.size());
+      List<ResourceId> dstFiles = new ArrayList<>(resultsToFinalFilenames.size());


Seems overkill - they are created right next to each other in code, and FileSystems.copy() already does that verification. I removed the size hints to make it a little simpler (preallocation probably doesn't matter here).

jkff · 2017-11-27T19:05:27Z

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

-    final PCollectionView<Integer> numShardsView;
+    PCollectionView<Integer> numShardsView =
+        (computeNumShards == null) ? null : input.apply(computeNumShards);
+    List<PCollectionView<Integer>> shardingSideInputs = numShardsView == null


jkff · 2017-11-27T19:06:30Z

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

-              "When finalizing a windowed write, should have set fixed sharding");
-        }
+        fixedNumShards = getFixedNumShards.apply(c);
+        checkState(fixedNumShards != null, "Windowed write should have set fixed sharding");


After #4124 they do - see #4137 for explanation.

jkff · 2017-11-27T19:13:20Z

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

-        c.output(
-            new FileResult<>(
-                writer.getOutputFile(), shardNumber, window, c.pane(), entry.getKey()));
+        int shard = c.element().getKey().getShardNumber();


jkff · 2017-11-27T19:15:39Z

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

+                  writer.getOutputFile(),
+                  shard,
+                  GlobalWindow.INSTANCE,
+                  PaneInfo.ON_TIME_AND_ONLY_FIRING,


Previously the code was structured differently, and the values passed in this particular codepath ended up being ignored. I consolidated things somewhat to handle much of windowed and unwindowed case the same way, and made the requirements more strict, in particular that window and pane have to be always set.

reuvenlax · 2017-11-30T04:01:44Z

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

          writer = writeOperation.createWriter();
-          int shardNumber =
-              shardNumberAssignment == ShardAssignment.ASSIGN_WHEN_WRITING


jkff wrote:
After #4124 they do - see #4137 for explanation.

We went through some trouble to ensure that windowed writes in batch do not require fixed sharding. 4124 fixes a bug that only affects triggered (generally streaming) writes, so it's unfortunate if it makes the batch case less usable. Is there any way to make windowed writes work in batch without requiring fixed sharding?

In #4137 I argued that this was never guaranteed to work. Or more like, it would only work if

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

Line 769 in 80ae930

.apply("FinalizeGroupByKey", GroupByKey.<Void, FileResult<DestinationT>>create())

fires exactly once and deterministically contains all the data - the code used "collection is bounded" as a proxy for that, but in reality this is not sufficient: Beam model does not prevent having a bounded collection triggering multiple times.

However, it seems that the only reason it doesn't work is that the order of things retrieved from a GBK is unstable - well, let's stabilize it then, with another reshuffle! I did that, and now this is supported - thanks for pushing :) Done.

jkff · 2017-12-01T00:04:14Z

Run Java PostCommit

…tions

jkff · 2017-12-04T20:26:50Z

Run Java PostCommit

jkff · 2017-12-05T22:28:40Z

PTAL?

chamikaramj

LGTM. Thanks.

jkff force-pushed the simplify-write-files branch from 1d78970 to 9cdeafb Compare November 17, 2017 23:41

reuvenlax reviewed Nov 22, 2017

View reviewed changes

jkff force-pushed the simplify-write-files branch from 9cdeafb to b3164b7 Compare November 23, 2017 00:34

jkff commented Nov 23, 2017

View reviewed changes

chamikaramj reviewed Nov 23, 2017

View reviewed changes

reuvenlax reviewed Nov 25, 2017

View reviewed changes

jkff commented Nov 27, 2017

View reviewed changes

reuvenlax reviewed Nov 30, 2017

View reviewed changes

jkff and others added 11 commits November 30, 2017 13:05

enforce fixed sharding

66fe469

Merges Writer.openWindowed/Unwindowed and removes result of close()

5c04a25

non-null window/pane in FileResult

05245dc

remove ShardAssignment

77b006f

consolidates windowed/unwindowed finalize fns somewhat

f1c85ec

Unifies windowed and unwindowed finalize.

cf06f7a

Refactors WriteFiles into sub-transforms

cc14687

Converts WriteFiles to AutoValue

eb9a1df

Makes checkstyle and findbugs happy

f4b5948

Renames spilled back to unwritten

d0bde4c

Fixes tests

0ae7ff0

jkff force-pushed the simplify-write-files branch from 8c56c07 to 6fc276f Compare December 1, 2017 00:03

Reintroduces dynamic sharding with windowed writes for bounded collec…

cc40f39

…tions

jkff force-pushed the simplify-write-files branch from 6fc276f to cc40f39 Compare December 4, 2017 20:26

chamikaramj reviewed Dec 6, 2017

View reviewed changes

asfgit closed this in 761ec1a Dec 6, 2017

jkff deleted the simplify-write-files branch December 6, 2017 00:30

		@@ -868,53 +934,10 @@ protected void finishWrite() throws Exception {}
		* id populated for the case of static sharding. In cases where the runner is dynamically

Many simplifications to WriteFiles #4145

Many simplifications to WriteFiles #4145

Conversation

jkff commented Nov 17, 2017

asfgit commented Nov 17, 2017

asfgit commented Nov 17, 2017

asfgit commented Nov 17, 2017

jkff commented Nov 20, 2017

reuvenlax commented Nov 21, 2017 via email

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

jkff left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj 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

jkff 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

jkff commented Dec 1, 2017

jkff commented Dec 4, 2017

jkff commented Dec 5, 2017

chamikaramj left a comment

Choose a reason for hiding this comment