python - Airflow :将 {{ ds }} 作为参数传递给 PostgresOperator

Question

我想使用执行日期作为我的 sql 文件的参数:

我试过了

dt = '{{ ds }}'

s3_to_redshift = PostgresOperator(
    task_id='s3_to_redshift',
    postgres_conn_id='redshift',
    sql='s3_to_redshift.sql',
    params={'file': dt},
    dag=dag
)

但它不起作用。

Answer 1

dt = '{{ ds }}'

不起作用，因为 Jinja( Airflow 中使用的模板引擎)不处理整个 Dag 定义文件。

对于每个 Operator 都有 Jinja 将处理的字段，它们是运算符本身定义的一部分。

在这种情况下，如果您扩展 PostgresOperator<，您可以使 params 字段(实际上称为 parameters，确保更改它)模板化 像这样:

class MyPostgresOperator(PostgresOperator):
    template_fields = ('sql','parameters')

现在你应该能够做到:

s3_to_redshift = MyPostgresOperator(
    task_id='s3_to_redshift',
    postgres_conn_id='redshift',
    sql='s3_to_redshift.sql',
    parameters={'file': '{{ ds }}'},
    dag=dag
)