关于向运行中的GCP Cloud Composer流水线及Dataproc工作流模板内作业传递参数的技术咨询

嘿，我来帮你梳理这两个GCP相关的问题，都是我实际工作中接触过的场景，应该能给你清晰的答案：

当然可以！Cloud Composer本质是托管的Airflow环境，给运行中的流水线（也就是DAG运行实例）传递参数有几种常用方式：

• Airflow UI手动触发：在Airflow UI找到目标DAG，点击"Trigger DAG w/ config"，在弹出的窗口里填入JSON格式的参数（比如{"input_path": "gs://my-bucket/data.csv"}），这些参数会被注入到DAG运行的上下文中，任务里可以通过dag_run.conf.get("input_path")来获取值。
• gcloud命令行触发：用命令行传递参数的示例如下：

  gcloud composer environments run my-composer-env --location us-central1 dags trigger -- my-dag-id --conf '{"input_path": "gs://my-bucket/data.csv"}'
  

• Airflow REST API触发：调用Airflow的DAG触发API时，在请求体的conf字段中传入参数JSON即可。

完全可行！具体实现需要两步配合：

1. 预先在Dataproc工作流模板中定义参数占位符：创建工作流模板时，在作业的配置里使用占位符（比如Spark作业的args里写{{input_path}}，或者Hive查询里用{{date}}）。
2. 在Cloud Composer的Operator中传递参数：使用Airflow的DataprocWorkflowTemplateOperator时，通过parameters字段传入键值对，这些值会自动替换模板中的对应占位符。示例代码如下：

   from airflow.providers.google.cloud.operators.dataproc import DataprocWorkflowTemplateOperator
   
   trigger_template = DataprocWorkflowTemplateOperator(
       task_id="trigger_dataproc_template",
       project_id="my-gcp-project",
       region="us-central1",
       template_id="my-dataproc-template",
       parameters={
           "input_path": "gs://my-bucket/data.csv",
           "date": "2024-05-20"
       }
   )
   

依据来源 #

这个功能是GCP官方支持的，根据官方文档说明：DataprocWorkflowTemplateOperator的parameters参数用于传递工作流模板的参数值，模板中定义的占位符会被这些值替换；同时Dataproc工作流模板本身支持参数化配置，允许在创建模板时预留变量，执行时动态传入值。

内容的提问来源于stack exchange，提问作者Snehil Singh