如何在JAVA开发的Dataflow作业内部获取当前作业的ID

首先明确：不需要额外调用谷歌云客户端库，Apache Beam 内置了现成的获取方法，直接通过 DataflowPipelineOptions 即可读取作业ID。

具体实现步骤 #

• 首先确认你的项目已经引入了Dataflow Runner的依赖，对应版本和你使用的Beam SDK版本对齐即可。
• 在流水线运行时的处理逻辑中（比如DoFn内部），将你拿到的PipelineOptions实例强转为DataflowPipelineOptions类型，调用getJobId()方法即可得到当前作业的运行ID。
  示例代码如下：

import org.apache.beam.runners.dataflow.DataflowPipelineOptions;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.DoFn;

// 示例DoFn，在处理元素时获取作业ID
public class GetJobIdFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) {
        PipelineOptions options = c.getPipelineOptions();
        DataflowPipelineOptions dataflowOpts = options.as(DataflowPipelineOptions.class);
        String currentJobId = dataflowOpts.getJobId();
        // 这里可以把currentJobId写入到你的BigQuery审计字段中
        c.output(currentJobId + "," + c.element());
    }
}

注意事项 #

• 禁止在流水线构造阶段调用该方法：本地构造流水线对象时，作业还未提交到Dataflow服务，此时getJobId()返回的是空值，必须在作业运行时的执行逻辑中调用才能拿到有效值。
• DirectRunner本地运行场景下getJobId()返回值为null，仅当流水线提交到Dataflow服务运行时才会返回正确的作业ID。
• 如果你需要在作业提交端（即你执行流水线提交命令的本地代码侧）提前拿到作业ID，可以在提交作业后通过DataflowPipelineJob实例获取，示例如下：

import org.apache.beam.runners.dataflow.DataflowPipelineJob;
import org.apache.beam.sdk.Pipeline;

// 构造流水线逻辑
Pipeline pipeline = Pipeline.create(options);
pipeline.apply(/*你的流水线处理逻辑*/);
// 提交作业
DataflowPipelineJob job = (DataflowPipelineJob) pipeline.run();
// 提交完成后即可拿到作业ID
String jobId = job.getJobId();

内容的提问来源于stack exchange，提问作者Shriyut Jha