python之Google Cloud Dataflow Python，检索作业 ID

我目前正在使用 Python 开发数据流模板，我想访问作业 ID 并使用它保存到特定的 Firestore 文档。

是否可以访问作业 ID？

我在文档中找不到与此相关的任何内容。

请您参考如下方法:

您可以通过从管道中调用 dataflow.projects().locations().jobs().list 来实现(请参阅下面的完整代码)。一种可能性是始终使用相同的作业名称调用模板，这是有道理的，否则作业前缀可以作为运行时参数传递。应用正则表达式解析作业列表以查看作业是否包含名称前缀，如果包含，则返回作业 ID。如果有多个，它只会返回最新的一个(当前正在运行的那个)。

在定义了 PROJECT 和 BUCKET 变量后，模板被暂存，其中:

python script.py \ 
    --runner DataflowRunner \ 
    --project $PROJECT \ 
    --staging_location gs://$BUCKET/staging \ 
    --temp_location gs://$BUCKET/temp \ 
    --template_location gs://$BUCKET/templates/retrieve_job_id 

然后，在执行模板作业时指定所需的作业名称(在我的例子中为 myjobprefix):

gcloud dataflow jobs run myjobprefix \ 
   --gcs-location gs://$BUCKET/templates/retrieve_job_id 

retrieve_job_id 函数将从作业中返回作业 ID，更改 job_prefix 以匹配给定的名称。

import argparse, logging, re 
from googleapiclient.discovery import build 
from oauth2client.client import GoogleCredentials 
import apache_beam as beam 
from apache_beam.options.pipeline_options import PipelineOptions 
from apache_beam.options.pipeline_options import SetupOptions 
 
 
def retrieve_job_id(element): 
  project = 'PROJECT_ID' 
  job_prefix = "myjobprefix" 
  location = 'us-central1' 
 
  logging.info("Looking for jobs with prefix {} in region {}...".format(job_prefix, location)) 
 
  try: 
    credentials = GoogleCredentials.get_application_default() 
    dataflow = build('dataflow', 'v1b3', credentials=credentials) 
 
    result = dataflow.projects().locations().jobs().list( 
      projectId=project, 
      location=location, 
    ).execute() 
 
    job_id = "none" 
 
    for job in result['jobs']: 
      if re.findall(r'' + re.escape(job_prefix) + '', job['name']): 
        job_id = job['id'] 
        break 
 
    logging.info("Job ID: {}".format(job_id)) 
    return job_id 
 
  except Exception as e: 
    logging.info("Error retrieving Job ID") 
    raise KeyError(e) 
 
 
def run(argv=None): 
  parser = argparse.ArgumentParser() 
  known_args, pipeline_args = parser.parse_known_args(argv) 
 
  pipeline_options = PipelineOptions(pipeline_args) 
  pipeline_options.view_as(SetupOptions).save_main_session = True 
 
  p = beam.Pipeline(options=pipeline_options) 
 
  init_data = (p 
               | 'Start' >> beam.Create(["Init pipeline"]) 
               | 'Retrieve Job ID' >> beam.FlatMap(retrieve_job_id)) 
 
  p.run() 
 
 
if __name__ == '__main__': 
  run()