如何将自定义MyPipelineOptions传递到Google Dataflow DOFN？

Question

java.lang.IllegalArgumentException:
com.xxx.MyProcessor,
@ProcessElement parseItem(PubsubMessage, MyPipelineOptions, OutputReceiver),
@ProcessElement parseItem(PubsubMessage, MyPipelineOptions, OutputReceiver),
parameter of type MyPipelineOptions at index 1:
MyPipelineOptions is not a valid context parameter.
Should be one of [BoundedWindow, RestrictionTracker<?, ?>]

如果我将MyPipelineOptions更改为PipelineOptions，错误就消失了，但是如果我试图在函数中强制转换回MyPipelineOptions，我会得到一个ClassCastException，所以我猜这不是正确的方法...知道如何将自定义选项类传递给元素处理器吗？

下面是代码结构：

import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;

public interface MyPipelineOptions extends DataflowPipelineOptions {
  ...
}

public class MyProcessor extends DoFn<PubsubMessage, String> {
  @ProcessElement
  public void parseItem(@Element PubsubMessage message, MyPipelineOptions po, OutputReceiver<String> out) throws Exception {
    ...
  }
}

注意文档只显示了一个非自定义PipelineOptions的示例：

PipelineOptions：当前管道的PipelineOptions始终可以通过将其作为参数添加到流程方法中来访问：

.of(new DoFn<String, String>() {
  public void processElement(
    @Element String word, PipelineOptions options) {

  }
 })

Answer 1

public class MyProcessor extends DoFn<PubsubMessage, String> {
  @ProcessElement
  public void parseItem(
    @Element PubsubMessage message,
    PipelineOptions po,
    OutputReceiver<String> out) throws Exception {

      MyPipelineOptions opts = po.as(MyPipelineOptions.class);
      ...
    }
  }
}