Airflow中ClickHouseOperator无法替换ds模板变量求助

核心解决步骤： #

• 修正调度配置：你当前DAG的schedule_interval=None，这会导致DAG无法自动触发，手动触发时execution_date（ds的来源）也可能不符合预期。如果是定期调度需求，改成对应调度表达式，比如每日调度：schedule_interval='@daily'。

• 用params传递变量更稳妥：直接在sql中写{{ds}}无效的话，通过params参数传递变量，再在sql里引用{{ params.exec_date }}：

insert_task = ClickhouseOperator(
    task_id="insert_task",
    ch_connection_id="clickhouse_db_connection",
    sql="""
           INSERT INTO tmp.target_tbl
            SELECT created_date,count() AS total_ride,
            count(DISTINCT(passengerId)) AS total_passenger 
            FROM tmp.source_tbl
            WHERE created_date >= '{{ params.exec_date }}'
            GROUP BY created_date
            """,
    params={"exec_date": "{{ds}}"},
    dag=dag,
)

• 确认Operator的模板支持：如果你使用的是apache-airflow-providers-clickhouse包的ClickHouseOperator，sql参数默认属于模板字段，支持Jinja渲染。如果是自定义旧版Operator，可能需要手动指定模板字段：

from airflow.providers.clickhouse.operators.clickhouse import ClickHouseOperator

class CustomClickHouseOperator(ClickHouseOperator):
    template_fields = ('sql',)

• 验证ds变量是否生成：可以加个PythonOperator打印ds值，确认execution_date是否正常生成：

from airflow.operators.python import PythonOperator

def print_ds(**context):
    print(f"当前ds值: {context['ds']}")

print_task = PythonOperator(
    task_id="print_ds",
    python_callable=print_ds,
    provide_context=True,
    dag=dag,
)

print_task >> insert_task

注意事项： #

• 不需要用转义的{{{{ds}}}}，正常的{{ds}}在支持模板的参数中会被正确渲染。
• 确保Airflow版本与clickhouse provider版本兼容，建议使用最新稳定版的provider包。

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