如何在Airflow的PostgresOperator中获取执行日期/逻辑日期？

实现步骤 #

Airflow的模板变量可直接在SQL文件中使用，无需context前缀，具体操作如下：

1. 修改SQL文件
   将some_query.sql中的硬编码日期替换为Airflow模板变量，注意根据数据库date字段类型调整格式：

SELECT *
FROM TABLE
WHERE date = '{{ logical_date.strftime("%Y-%m-%d") }}'

若字段是带时间戳的类型，可使用原生ISO格式：

WHERE date = '{{ logical_date.isoformat() }}'

2. 确认PostgresOperator的模板支持
   PostgresOperator默认已启用模板解析，你的现有代码无需额外修改即可生效：

task = PostgresOperator(
    task_id="task_id",
    postgres_conn_id="local_conn",
    sql="sql/some_query.sql"
)

3. 可选：通过params自定义参数传递
   如果需要更灵活的参数管理，可在Operator中定义params，再在SQL中引用：

task = PostgresOperator(
    task_id="task_id",
    postgres_conn_id="local_conn",
    sql="sql/some_query.sql",
    params={"target_date": "{{ logical_date.strftime('%Y-%m-%d') }}"}
)

对应的SQL写法：

SELECT *
FROM TABLE
WHERE date = '{{ params.target_date }}'

基于日期调整任务行为是否为最佳实践？ #

• 这是Airflow的核心最佳实践之一：Airflow专为周期性调度任务设计，基于logical_date（或execution_date）过滤数据、生成报表等是ETL/ELT工作流的典型模式，能保证任务的幂等性——相同logical_date执行时，输出结果一致。
• 关键注意事项：
  • 优先使用logical_date而非系统当前时间，即使任务延迟执行，也能确保数据范围符合调度周期的要求。
  • 确保日期格式与数据库字段类型匹配，避免隐式转换引发的错误或性能损耗。
  • 跨时区场景下，需对齐Airflow与数据库的时区设置，或在模板中显式指定时区转换规则。

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