Как применить DoFn PTransform к PCollectionTuple в Apache Beam

Question

Я пытаюсь применить PTransform к PCollectionTuple, но не могу понять, почему компилятор жалуется.

Я хочу сделать это, чтобы абстрагироваться в один PTransform для нескольких необходимых шагов для объединения некоторых строк CSV (каждая PCollection в PCollectionTuple содержит строки CSV для объединения), и у меня проблема не в самом соединении, а в том, как применить PTransform к PCollectionTuple.

Это мой код:

static class JoinCsvLines extends DoFn<PCollectionTuple, String[]> {
        @ProcessElement
        public void processElement(ProcessContext context) {
            PCollectionTuple element = context.element();
            // TODO: Implement the output
        }
    }

И я вызываю PTransform следующим образом:

TupleTag<String[]> tag1 = new TupleTag<>();
TupleTag<String[]> tag2 = new TupleTag<>();
PCollectionTuple toJoin = PCollectionTuple.of(tag1, csvLines1).and(tag2, csvLines2);

// Can't compile this line
PCollection<String[]> joinedLines = toJoin.apply("JoinLines", ParDo.of(new JoinCsvLines()));

IntelliJ IDEA выводит следующее, когда я нахожусь над строкой, которая не компилируется:

Required type:
PTransform
<? super PCollectionTuple,
OutputT>
Provided:
SingleOutput
<PCollectionTuple,
String[]>
reason: no instance(s) of type variable(s) InputT exist so that PCollectionTuple conforms to PCollection<? extends InputT>

Как я могу применить PTransform к PCollectionTuple?

Answer 1

DoFn<PCollectionTuple, String[]> означает, что вы хотите применить DoFn для каждой записи, поэтому вы не должны использовать PCollectionTuple в качестве типа ввода. Вместо этого вы должны использовать тип ваших "csvLines1" и "csvLines2".

Если вы собираетесь объединить две PCollections, вы можете проверить преобразование Flatten: https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Flatten.java#L41