本文介紹了一個數據流作業內的并行管道的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！
問題描述
我要在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個未連接的”管道”。
這篇關于一個數據流作業內的并行管道的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
