Apache Beam: Flattening PCollection to PCollection

Question

Say we have some nested list:

["a", "b"]
["c", "d"]

And we can easily do the flap map in Stream API like this:

Stream
        .of(List.of("a", "b"), List.of("c", "d"))
        .flatMap(List::stream)
        .forEach(System.out::println);

But doing it with "FlatMapElements", it was quite a mess:

Pipeline pipeline = Pipeline.create();
pipeline.apply(Create.of(List.of(List.of("a", "b"), List.of("c", "d"))))
        .apply(FlatMapElements.into(TypeDescriptor.of(String.class)).via(list -> list))
        .apply(ParDo.of(new SomeOutputFunction()));

Can we do anything else better with the flat map function?
A simple flatmap job should not be that complicated so I think I am missing something.
I cannot even replace .via(list -> list) to .via(Function.identity()) due to the type inference problem.

user_15199880 · Answer

".apply(Flatten.iterables())" it will convert "PCollection" to "PCollection"

user_3175889 · Answer

Please refer to the Apache Beam programming guide https://beam.apache.org/documentation/programming-guide/#flatten:

PCollectionList collections = PCollectionList.of(pc1).and(pc2).and(pc3);

PCollection merged = collections.apply(Flatten.pCollections());

Apache Beam: Flattening PCollection<List<Foo>> to PCollection<Foo>

2 Answers