[BEAM-9346] TFRecordIO inefficient read from sideinput causing pipeline to be slow

• Type: Improvement
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: Not applicable
• Component/s: sdk-java-core
• Labels: Done dataflow easyfix performance
• Assignee: Piotr Szuberski
• Reporter: Ban Piao
• Created: 20/Feb/20 23:27
• Updated: 06/Oct/20 21:55
• Resolved: 16/Mar/20 09:09

Description

In TFRecordIO, Reify.viewInGlobalWindow(input.apply(View.asList()), ListCoder.of(resultCoder)) is an inefficient way of reading large set of side input.

Pipeline can be sped up significantly by combinging the PCollection<ResultT> to a single element PCollection<List<ResultT>>.

Sample code: 

 https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L412 from
```
return input
.getPipeline()
.apply(Reify.viewInGlobalWindow(input.apply(View.asList()), ListCoder.of(resultCoder)));
```

to

```
return input.apply("ToList", Combine.globally(new ToListCombineFn<>()));
```
where ToListCombineFn is defined as
```
public static class ToListCombineFn<ResultT> extends CombineFn<ResultT, List<ResultT>, List<ResultT>> {

@Override
public List<ResultT> createAccumulator()

{ return new ArrayList<>(); }

@Override
public List<ResultT> addInput(List<ResultT> mutableAccumulator, ResultT input)

{ mutableAccumulator.add(input); return mutableAccumulator; }

@Override
public List<ResultT> mergeAccumulators(Iterable<List<ResultT>> accumulators) {
Iterator<List<ResultT>> iter = accumulators.iterator();
if (!iter.hasNext())

{ return new ArrayList<>(); }

List<ResultT> merged = iter.next();
while (iter.hasNext())

{ merged.addAll(iter.next()); }

return merged;
}

@Override
public List<ResultT> extractOutput(List<ResultT> accumulator)

{ return accumulator; }

}
```

Issue Links

• causes: BEAM-9563 TFRecordIO inefficient read from sideinput causing pipeline to be slow - fix
• links to: GitHub Pull Request #11122