Postgres Operator 的气流参数

Question

我正在尝试将执行日期作为运行时参数传递给 postgres 运算符

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

task = MyPostgresOperator(
  task_id='test_date',
  postgres_conn_id='redshift',
  sql="test_file.sql",
  parameters={'crunch_date':'{{ ds }}'},
  dag=dag
)

然后我尝试在sql查询中使用这个参数来接受dag传递的值

select 
{{ crunch_date }} as test1,

dag 正确发送参数，但是查询只是采用空值而不是传递的执行日期。有没有办法让带有 redshift 的 postgresql 接受这个参数的正确值？

Answer 1

您必须按如下方式更新您的 sql 查询：

select 
{{ ds }} as test1,

您将无法在另一个中使用一个模板化字段。如果要在任务中传递参数并在 Jinja 模板中使用它，请使用 params 参数。

更新：

但请注意params 不是模板字段。如果你模板化它，它不会渲染，因为嵌套模板不起作用。

task = MyPostgresOperator(
  task_id='test_date',
  postgres_conn_id='redshift',
  sql="test_file.sql",
  params={'textstring':'abc'},
  dag=dag
)

test_file.sql 在哪里：

select 
{{ params.textstring }} as test1,

查看https://medium.com/datareply/airflow-lesser-known-tips-tricks-and-best-practises-cf4d4a90f8f 中的第 4 点以了解有关params 的更多信息。

Answer 2

您可以在查询字符串中使用气流宏 - 需要将其传递给 redshift。

例子：

PostgresOperator(task_id="run_on_redshift",
                 dag=dag,
                 postgres_conn_id=REDSHIFT_CONN_ID,
                 sql="""
                        UNLOAD ('select * from abc.xyz') TO 's3://path/{{ds}}/' iam_role 's3_iam_role' DELIMITER AS '^' ALLOWOVERWRITE addquotes ESCAPE HEADER parallel off;
                     """
                 )