如何在Cloud Composer中并行执行同一PythonOperator多任务？

完全可以实现需求，以下是两种可行方案：

• 推荐方案：使用动态任务映射（Dynamic Task Mapping）
  这是Airflow 2.3+版本支持的原生功能，Cloud Composer 2.0+环境均兼容，能简洁生成并行子任务：

  with DAG(
      dag_id=config['dag_id'],
      schedule_interval=config['schedule_interval'],
      default_args=default_args
  ) as dag:
      # 通过partial定义固定参数，expand触发10次并行执行
      generate_data_task = PythonOperator.partial(
          task_id='generate_dummy_data',
          python_callable=generate_data,
      ).expand(
          # 传递长度为10的列表作为映射触发条件，参数可在函数中忽略
          dummy_param=list(range(10))
      )
  

• 备选方案：循环创建独立任务
  通过循环生成10个拥有唯一task_id的PythonOperator任务，实现并行：

  with DAG(
      dag_id=config['dag_id'],
      schedule_interval=config['schedule_interval'],
      default_args=default_args
  ) as dag:
      # 循环生成10个并行任务
      for idx in range(10):
          PythonOperator(
              task_id=f'generate_dummy_data_{idx}',
              python_callable=generate_data,
              dag=dag
          )
  

关于concurrency和task_concurrency参数无效的原因 #

• concurrency是DAG级别的全局并发控制，限制整个DAG同时运行的任务总数；
• task_concurrency是单个task_id的并发数，但Airflow不会重复执行同一个task_id的任务实例，因此仅设置这两个参数无法实现同一任务的多次并行执行，必须生成多个独立的任务实例。

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