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(气流中使用的模板引擎)不处理整个Dag定义文件.

对于每个Operator领域,Jinja将处理这些领域,这些领域是运营商本身的定义的一部分.

在这种情况下,如果扩展如下所示,您可以使该params字段(实际上被调用parameters,确保更改它)模板化PostgresOperator:

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
)