如何在PostGresOperator气流中传递我的.sql文件中的参数?

Question

我有sql文件，我想用PostGresOperator传递参数到sql文件。

"""select * from table_{} where id > ID """.format(mytable,myID)

我postGresOperator

mport_redshift_table = PostgresOperator(
task_id='copy_data_from_redshift_{}'.format(country),
postgres_conn_id='postgres_default',
sql="""
select * from table_{} where id > {}
""".format(mytable,myID)

我怎么能做同样的，在我的。sql文件中传递我的参数，仍然使用。format(mytable,myID) ?

，以便我可以将它们传递到我引用的。sql文件中。

Answer 1

正如PostgresOperator的入门指南中所解释的，您可以将SQL放在dag目录下的子文件夹中的文件中:

-- dags/sql/birth_date.sql
SELECT * FROM pet WHERE birth_date BETWEEN SYMMETRIC {{ params.begin_date }} AND {{ params.end_date }};

使用params传递将在文件中的SQL中呈现的键/值对:

get_birth_date = PostgresOperator(
task_id="get_birth_date",
postgres_conn_id="postgres_default",
sql="sql/birth_date.sql",
params={"begin_date": "2020-01-01", "end_date": "2020-12-31"},
)

编辑

如果你想内联，不使用文件，只需使用任何字符串插值机制:

sql="""select * from table_{} where id > {} """.format(mytable,myID)

或

sql=f"""select * from table_{table_name} where id > {myID} """

或者如果您想使用jinja，利用任何默认变量，例如触发DAG时提供的参数，您可以这样做:

sql=f"""select * from table_{{ params.param1 }} where id > {myID} """