如何在Java语言中向DoFn传递和引用多个侧输入

import org.apache.beam.sdk.transforms.DoFn;

public class MyDoFn extends DoFn<InputType, OutputType> {
  // DoFn逻辑实现
}

import org.apache.beam.sdk.transforms.DoFn;

public class MyDoFn extends DoFn<InputType, OutputType> {
  @ProcessElement
  public void processElement(ProcessContext c) {
    // 访问主输入元素
    InputType mainInput = c.element();
    
    // 访问侧输入1
    PCollectionView<SideInputType1> sideInput1 = c.sideInput(sideInput1View);
    SideInputType1 sideInputData1 = c.sideInput(sideInput1);
    
    // 访问侧输入2
    PCollectionView<SideInputType2> sideInput2 = c.sideInput(sideInput2View);
    SideInputType2 sideInputData2 = c.sideInput(sideInput2);
    
    // 执行逻辑处理，并输出结果
    // ...
    c.output(output);
  }
}

import org.apache.beam.sdk.transforms.View;

// 创建侧输入PCollectionView
PCollectionView<SideInputType1> sideInput1View = mainInputPCollection.apply(View.asSingleton());

// 将侧输入与DoFn绑定
PCollection<OutputType> output = mainInputPCollection.apply(ParDo.of(new MyDoFn())
    .withSideInputs(sideInput1View, sideInput2View));

// 提供侧输入的值
SideInputType1 sideInputData1 = ...;
SideInputType2 sideInputData2 = ...;
TupleTag<SideInputType1> sideInput1Tag = new TupleTag<>();
TupleTag<SideInputType2> sideInput2Tag = new TupleTag<>();

// 运行Pipeline并传递侧输入的值
PipelineResult result = pipeline.run();
result.waitUntilFinish();