用Java在Google Dataflow1.9.0中多次使用DoFn

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;

public class DataflowExample {
  public static void main(String[] args) {
    // 创建PipelineOptions对象
    PipelineOptions options = PipelineOptionsFactory.create();

    // 创建Pipeline对象
    Pipeline pipeline = Pipeline.create(options);

    // 从文本文件读取数据
    pipeline.apply(TextIO.read().from("input.txt"))
        // 第一个DoFn，将每行字符串拆分为单词
        .apply(ParDo.of(new SplitWordsFn()))
        // 第二个DoFn，将单词转换为大写
        .apply(ParDo.of(new UppercaseFn()))
        // 第三个DoFn，输出结果
        .apply(ParDo.of(new OutputFn()));

    // 运行Pipeline
    pipeline.run().waitUntilFinish();
  }

  // 第一个DoFn，将每行字符串拆分为单词
  public static class SplitWordsFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      String[] words = c.element().split(" ");
      for (String word : words) {
        c.output(word);
      }
    }
  }

  // 第二个DoFn，将单词转换为大写
  public static class UppercaseFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      String word = c.element().toUpperCase();
      c.output(word);
    }
  }

  // 第三个DoFn，输出结果
  public static class OutputFn extends DoFn<String, Void> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      System.out.println(c.element());
    }
  }
}