Airflow RedshiftToS3Operator：如何使用parameters传递查询参数

你当前的问题是混淆了SQL参数绑定和Airflow Jinja模板渲染的两种用法，导致参数无法生效。下面给两种可行的解决方法：

方法一：使用SQL参数绑定（对应Operator的parameters参数） #

RedshiftToS3Operator的parameters参数是用于SQL语句的参数化查询（防止SQL注入），需要在SQL里用占位符匹配参数，而非直接用Jinja变量语法。

修改mycode.sql： #

用Redshift支持的命名占位符格式%(参数名)s：

SELECT * FROM my_table 
WHERE date_start = CURRENT_DATE - INTERVAL '%(day_as_parameters)s days'

保持Operator代码不变： #

unload_data_to_s3 = RedshiftToS3Operator(
        task_id="unload_data_to_s3",
        s3_bucket=S3_BUCKET,
        s3_key=S3_KEY,
        select_query="mycode.sql",
        unload_options=["CSV", "ALLOWOVERWRITE", "DELIMITER ','", "PARALLEL OFF", "HEADER"],
        table_as_file_name=False,
        parameters={"day_as_parameters": 10}
    )

方法二：使用Airflow Jinja模板渲染（无需parameters参数） #

如果要在SQL里使用data_interval_end这类Airflow内置模板变量，同时传入自定义参数，直接将参数作为Operator的属性传入即可（select_query属于RedshiftToS3Operator的模板渲染字段）。

修改Operator代码： #

把自定义参数直接作为Operator的关键字参数传入：

unload_data_to_s3 = RedshiftToS3Operator(
        task_id="unload_data_to_s3",
        s3_bucket=S3_BUCKET,
        s3_key=S3_KEY,
        select_query="mycode.sql",
        unload_options=["CSV", "ALLOWOVERWRITE", "DELIMITER ','", "PARALLEL OFF", "HEADER"],
        table_as_file_name=False,
        day_as_parameters=10  # 直接传入模板变量
    )

保持mycode.sql不变： #

SELECT * FROM my_table 
WHERE date_start = {{ data_interval_end.subtract(days=day_as_parameters) }}

注意事项： #

• 方法一适合纯SQL层面的参数替换，和Airflow模板无关；
• 方法二适合需要结合Airflow调度上下文（如data_interval_end）的场景，属于模板渲染逻辑。

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