对一个输出标签应用变换答案

【问题标题】：Applying a transform to one output tag对一个输出标签应用变换
【发布时间】：2018-12-11 12:23:50
【问题描述】：

我想我有一个产生两个输出的函数（如果我错了，请纠正我）：

PCollection<String> words = ...;

final TupleTag<String> shortWordsTag = new TupleTag<String>(){};

PCollectionTuple results =
     words.apply(
         ParDo
         .of(new DoFn<String, String>() {
             @ProcessElement
             public void processElement(ProcessContext context) {
                 String word = context.element();
                 if (word.length() < 5) {
                     context.output(shortWordsTag, word);
                 } else {
                     context.output(word);
             }

现在我想调用另一个函数，但只应用其中一个输出。像这样的：

results.apply(
    ParDo
    .of(new DoFn<String, String>() {
        @ProcessElement
        public void processElement(ProcessContext context) {
            String word = context.element();
            // do stuff, but should only have words with length < 5 here
    }
)

我可以看到一些使用withOutputTags 的示例，但这种方法似乎需要多个标签（一个标签和一个标签列表），我不确定如何在我的场景中使用它。

我如何指定我的results.apply 仅被输出到shortWordsTag 标签的数据调用？

【问题讨论】：

标签： java google-cloud-dataflow apache-beam

【解决方案1】：

正如您所提到的，在 Apache Beam 的单个转换中处理多个输出的正确方法确实是使用 PCollectionTuple 和 withOutputTags。

在 Apache Beam 文档中，您可以找到一些非常好的示例，说明如何设置具有多个输出的转换，并为每个输出使用不同的标签：

此外，如果您访问上面第二个链接中的第 4.5.2 节，您将找到有关如何在 DoFn 中向多个输出发射的示例。简而言之，使用您共享的核心代码，您需要执行以下操作：

PCollectionTuple results = [...].withOutputTags(MAIN_TAG, LIST_OF_ADDITIONAL_TAGS);

results.get(YOUR_DESIRED_TAG).apply(...);

调用get( ) method on a PCollectionTuple 将返回与您将在方法内传递的TupleTag 关联的PCollection。

【讨论】：