气流 - 重用任务答案

【问题标题】：airflow - reuse a task气流 - 重用任务
【发布时间】：2021-01-25 02:56:33
【问题描述】：

我正在将一些表从 PostgreSQL 导出到 GCS。为了使它看起来简单，我创建了如下所示的 dag。导出日期是这样的。

from airflow.models import DAG
from airflow.contrib.operators.postgres_to_gcs_operator import  PostgresToGoogleCloudStorageOperator

def sub_dag_export(parent_dag_name, child_dag_name, args, export_suffix):
    dag = DAG(
      '%s.%s' % (parent_dag_name, child_dag_name),
      default_args=args,
      start_date=args['start_date'],
      max_active_runs=1,
    )
    export_tbl1 = PostgresToGoogleCloudStorageOperator(
        task_id='export_tbl1',
        postgres_conn_id='cloudsqlpg',
        google_cloud_storage_conn_id='gcsconn',
        sql='SELECT * FROM tbl1',
        export_format='csv',
        field_delimiter='|',
        bucket='dsrestoretest',
        filename='file/export_tbl1/tbl1_{}.csv',
        schema_filename='file/schema/tbl1.json',
        dag=dag)

        export_tbl1 = PostgresToGoogleCloudStorageOperator(
        task_id='export_tbl2',
        postgres_conn_id='cloudsqlpg',
        google_cloud_storage_conn_id='gcsconn',
        sql='SELECT * FROM tbl2',
        export_format='csv',
        field_delimiter='|',
        bucket='dsrestoretest',
        filename='file/export_tbl1/tbl2_{}.csv',
        schema_filename='file/schema/tbl2.json',
        dag=dag)

任务 1 和 2 都在做同样的工作，所以我想为所有表重用我的 export1 任务。但它不应该改变流程。 (Start --> export table1 --> table2 -->table3 --end)，因为由于某些原因，如果任务失败，我需要从失败的地方重新运行任务。所以即使我要使用单个任务，DAG 图也应该是一样的。

我看到有一种方法（from this link），但我仍然无法完全理解这一点。

【问题讨论】：

标签： airflow orchestration

【解决方案1】：

您只需将公共代码提取到一个函数中并让它为您创建运算符实例。

def pg_table_to_gcs(table_name: str) -> PostgresToGoogleCloudStorageOperator:
  return PostgresToGoogleCloudStorageOperator(
      task_id=f"export_{table_name}",
      postgres_conn_id="cloudsqlpg",
      google_cloud_storage_conn_id="gcsconn",
      sql=f"SELECT * FROM {table_name}",
      export_format="csv",
      field_delimiter="|",
      bucket="dsrestoretest",
      filename=f"file/export_{table_name}/{table_name}.csv",
      schema_filename=f"file/schema/{table_name}.json",
      dag=dag)

tables = ["table0", "table1", "table2"]

with DAG(dag_id="kube_example", default_args=default_args) as dag:
    reduce(lambda t0, t1: t0 >> t1, [pg_table_to_gcs(table, dag) for table in table_names])

【讨论】：

@bhuvanesh 使用表名列表更新了 dag 创建