<commit_before># Dataflow で合流する例

sink はそれぞれの出力先に適したレート制御のロジックなどが実装されているので [multi-source-multi-sink](/dataflow/example/multi-source-multi-sink/README.md) に書かれているもののように、
同じ sink に対して複数の PCollection を書き込む場合は sink を合流したほうが効率がよくなる場合がある。

合流するには `PCollectionList#apply` に Flatten を渡すなどすると良い。

```java
public static void main(String[] args) throws CannotProvideCoderException {
    DatastoreToDatastoreOptions options =
            PipelineOptionsFactory.fromArgs(args).withValidation()
                    .as(DatastoreToDatastoreOptions.class);
    String[] kinds = options.getInputKinds().split(",");
    Pipeline p = Pipeline.create(options);
    PCollectionList<Entity> pcs = Arrays.stream(kinds).map(kind -> {
                KindExpression kindExpression = KindExpression.newBuilder().setName(kind).build();
                Query getKindQuery = Query.newBuilder().addKind(kindExpression).build();
        PCollection<Entity> pc = p.apply(kind, DatastoreIO.v1().read().withProjectId(options.getInputProjectId()).withQuery(getKindQuery))
                .apply(new EntityMigration());
        return pc;
            }
    ).collect(Collector.<PCollection<Entity>, PCollectionList<Entity>>of(() -> PCollectionList.empty(p),
            PCollectionList::and,
            (pcs1, pcs2) -> pcs1.and(pcs2.getAll())));
    pcs.apply(Flatten.pCollections()).setCoder(p.getCoderRegistry().getCoder(Entity.class))
            .apply(DatastoreIO.v1().write().withProjectId(options.getOutputProjectId()));
    p.run();
}
```<commit_msg>PCollectionList を collect で作るのをやめる<commit_after># Dataflow で合流する例

sink はそれぞれの出力先に適したレート制御のロジックなどが実装されているので [multi-source-multi-sink](/dataflow/example/multi-source-multi-sink/README.md) に書かれているもののように、
同じ sink に対して複数の PCollection を書き込む場合は sink を合流したほうが効率がよくなる場合がある。

合流するには `PCollectionList#apply` に Flatten を渡すなどすると良い。

```java
public static void main(String[] args) {
    DatastoreToDatastoreOptions options =
            PipelineOptionsFactory.fromArgs(args).withValidation()
                    .as(DatastoreToDatastoreOptions.class);
    String[] kinds = options.getInputKinds().split(",");
    Pipeline p = Pipeline.create(options);
    List<PCollection<Entity>> listOfPCollection = Arrays.stream(kinds).map(kind -> {
                KindExpression kindExpression = KindExpression.newBuilder().setName(kind).build();
                Query getKindQuery = Query.newBuilder().addKind(kindExpression).build();
                return p.apply(kind, DatastoreIO.v1().read().withProjectId(options.getInputProjectId()).withQuery(getKindQuery))
                        .apply(new EntityMigration());
            }
    ).collect(Collectors.toList());
    PCollectionList.of(listOfPCollection)
            .apply(Flatten.pCollections())
            .apply(DatastoreIO.v1().write().withProjectId(options.getOutputProjectId()));
    p.run();
}
```
{'fraction_non_alphanumeric': 0.12858141160027953, 'fraction_numerical': 0.002795248078266946, 'mean_word_length': 3.844331641285956, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7535423', 'n_tokens_mistral': 1019, 'n_tokens_neox': 980, 'n_words': 121}