java - 将 TupleTag 传递给 DoFn 方法

Question

我正在尝试从 DoFn 方法获得两个输出，以下是 Apache Beam programming guide 的示例

基本上在示例中你传递一个 TupleTag 然后指定输出到哪里，这对我有用问题是我在 ParDo 中调用了一个外部方法，不知道如何传递这个 TupleTag，这是我的代码:

PCollectionTuple processedData = pubEv
  .apply("Processing", ParDo.of(new HandleEv())
      .withOutputTags(mainData, TupleTagList.of(failedData)));

HandleEv 方法:

static class HandleEv extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) throws Exception {
      c.output("test")
      c.output(failedData,"failed")
    }
}

我得到的错误是找不到符号，因为无法从 HandleEv 访问 failedData，我尝试在类(class)开始时声明 failedData，但两者都不起作用。

非常感谢

Answer 1

您可以像将值传递给任何其他对象一样执行此操作 -- 将其作为参数传递给 HandleEv 的构造函数并将其存储在一个字段中:

static class HandleEv extends DoFn<String, String> {
  private final TupleTag<String> failedData;
  public HandleEv(TupleTag<String> failedData) {
    this.failedData = failedData;
  }

  @ProcessElement
  public void processElement(ProcessContext c) throws Exception {
    c.output("test")
    c.output(failedData,"failed")
  }
}

然后像这样使用它:

PCollectionTuple processedData = pubEv
  .apply("Processing", ParDo.of(new HandleEv(failedData))
      .withOutputTags(mainData, TupleTagList.of(failedData)));