Airflow的PostgresOperator如何向引用的.sql文件传递参数

解决方案 #

有两种常用实现方式，你可以根据需求选择： #

方式1：沿用原有format逻辑（和你现有写法习惯一致） #

1. 先编写外部SQL模板文件（比如命名为 query_template.sql），占位符格式和你之前写在代码里的保持一致：

select * from table_{} where id > {}

也可以用命名占位符降低传参顺序出错的概率：

select * from table_{mytable} where id > {myID}

2. 在DAG代码中先读取SQL文件内容，再调用format方法传参即可，示例代码：

# 先读取外部SQL文件内容
with open('/path/to/your/query_template.sql', 'r') as f:
    sql_content = f.read()

import_redshift_table = PostgresOperator(
    task_id='copy_data_from_redshift_{}'.format(country),
    postgres_conn_id='postgres_default',
    # 直接对读取到的SQL内容调用format
    sql=sql_content.format(mytable, myID)
    # 如果用的是命名占位符，传参写法为：
    # sql=sql_content.format(mytable=mytable, myID=myID)
)

方式2：使用PostgresOperator原生parameters参数（更安全，推荐） #

这种方式是官方推荐的参数化查询写法，可以避免SQL注入风险，不需要手动调用format：

1. 外部SQL文件（比如命名为 query_param.sql）中用命名占位符：

select * from table_%(mytable)s where id > %(myID)s

2. DAG代码中直接指定SQL文件路径，通过parameters参数传参即可：

import_redshift_table = PostgresOperator(
    task_id='copy_data_from_redshift_{}'.format(country),
    postgres_conn_id='postgres_default',
    # 直接填外部SQL文件的路径
    sql='/path/to/your/query_param.sql',
    # 通过parameters传参
    parameters={
        'mytable': mytable,
        'myID': myID
    }
)

注意：如果SQL文件路径是相对路径，需要确认和DAG文件的相对位置符合Airflow的路径解析规则。

内容的提问来源于stack exchange，提问作者Aditya Verma