Skip to content

[BEAM-26359] Avoid expensive AsList in FileWrites. #26360

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jrmccluskey merged 2 commits into from
Apr 20, 2023
Merged

[BEAM-26359] Avoid expensive AsList in FileWrites. #26360

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
52766ff
[BEAM-26289] Use iterable side inputs rather than list side inputs in…
robertwb Apr 14, 2023
e96b774
Implement file write fix in update-compatible way.
robertwb Apr 14, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Objects;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ArrayListMultimap;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Multimap;
Expand Down Expand Up @@ -462,7 +463,24 @@ public PCollection<List<ResultT>> expand(PCollection<ResultT> input) {
// iterable to finalize if there are no results.
return input
.getPipeline()
.apply(Reify.viewInGlobalWindow(input.apply(View.asList()), ListCoder.of(resultCoder)));
.apply(
"AsPossiblyEmptyList",
Reify.viewInGlobalWindow(
// Insert a reshuffle before taking the view to consolidate the (typically)
// one-output-per-bundle writes.
// This avoids producing a huge number of tiny files in the case that side
// inputs are materialized to disk bundle-by-bundle.
input.apply("Consolidate", Reshuffle.viaRandomKey()).apply(View.asIterable()),
IterableCoder.of(resultCoder)))
// View.asIterable() can be (significantly) cheaper than View.asList(), as it does not
// create a backing indexable view, but we must return a list to maintain update
// compatibility for consumers that are shared between this path and the streaming one.
.apply(
"IterableToList",
MapElements.via(
new SimpleFunction<Iterable<ResultT>, List<ResultT>>(
x -> ImmutableList.copyOf(x)) {}))
.setCoder(ListCoder.of(resultCoder));
}
}
}
Expand Down