一个数据流作业内的并行管道

我要在GCP上的一个数据流作业内运行两个并行管道。我已经创建了一个管道，并且工作正常，但我希望在不创建另一个作业的情况下创建另一个管道。

我搜索了这么多答案，但没有找到任何代码示例：(

如果我这样运行它，它不工作：

pipe1.run();
pipe2.run();

显示"已有活动作业名称...如果要提交第二个作业，请尝试使用--jobName重新设置其他名称"

---

解决方案

您可以将其他输入应用于管道，这将导致一个作业中的单独管道。例如：

public class ExamplePipeline {

public static void main(String[] args) {
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
    options.setRunner(DirectRunner.class);

    Pipeline pipeline = Pipeline.create(options);

    PCollection<String> linesForPipelineOne = pipeline.apply(Create.of("A1", "B1"));
    PCollection<String> linesToWriteFromPipelineOne = linesForPipelineOne.apply("Pipeline 1 transform",
            ParDo.of(new DoFn<String, String>() {

        @ProcessElement
        public void processElement(ProcessContext c) {
            System.out.println("Pipeline one:" + c.element());
            c.output(c.element() + " extra message.");
        }

    }));
    linesToWriteFromPipelineOne.apply((TextIO.write().to("file.txt")));

    PCollection<String> linesForPipelineTwo = pipeline.apply(Create.of("A2", "B2"));
    linesForPipelineTwo.apply("Pipeline 2 transoform",
            ParDo.of(new DoFn<String, String>() {

        @ProcessElement
        public void processElement(ProcessContext c) {
            System.out.println("Pipeline two:" + c.element());
        }

    }));

    pipeline.run();
}

如您所见，您还可以将两个(或更多)独立的PBegin应用于具有多个PDone/接收器的管道。在此示例中，"pipeline 1"将输出转储并写入文件，"pipeline 2"仅将其转储到屏幕。

如果您在GCP上使用DataflowRunner运行此命令，则图形用户界面将显示2个未连接的"管道"。