如何获取GCS_Text_to_Cloud_PubSub模板用于Airflow的DataflowTemplatedJobStartOperator

一、获取并复制官方模板到你的GCS存储桶 #

Google官方的Dataflow预定义模板存放在公共GCS桶gs://dataflow-templates/中，你可以通过以下步骤将GCS_Text_to_Cloud_PubSub模板复制到自己的存储桶：

• 确认模板的官方路径：最新版模板路径为gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub，若需要指定旧版本，可将latest替换为具体版本号（如2023-09-12_RC00）
• 使用gcloud命令行工具复制模板到你的桶：

  gcloud storage cp gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub gs://your-custom-bucket/dataflow-templates/
  

  替换your-custom-bucket为你的GCS存储桶名称，路径可根据自己的需求调整。

二、Airflow中配置DataflowTemplatedJobStartOperator #

复制完成后，直接在Operator中指向自己桶内的模板路径即可，示例代码如下：

from airflow.providers.google.cloud.operators.dataflow import DataflowTemplatedJobStartOperator

gcs_to_pubsub_task = DataflowTemplatedJobStartOperator(
    task_id="gcs_text_to_pubsub",
    template="gs://your-custom-bucket/dataflow-templates/GCS_Text_to_Cloud_PubSub",
    job_name="gcs-to-pubsub-{{ ds_nodash }}",  # 动态生成唯一Job名
    parameters={
        "inputFilePattern": "gs://your-input-bucket/*.txt",  # 输入文本文件路径
        "outputTopic": "projects/your-project-id/topics/your-target-topic"  # 目标Pub/Sub主题
    },
    location="us-central1",  # 根据你的需求指定Dataflow运行区域
    project_id="your-project-id"
)

额外提示 #

其实无需复制模板到自己的桶，你可以直接使用官方公共桶的模板路径作为template参数的值，比如：

template="gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub"

这样可以省去维护模板版本的麻烦，Google会自动维护latest指向最新稳定版。

内容的提问来源于stack exchange，提问作者Debpriya Seal