使用 Jinja 模板

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

什么是模板？

  • 一种预定义格式，使 Airflow 可自动替换信息
  • 常为纯文本/文本文件
  • 便于访问与复用

 

带占位符的模板示意图，用于替换值

Python 中的 Apache Airflow 入门

什么是 Jinja？

  • 简单的基于文本的模板语言
  • 不仅用于 Airflow，许多工具都在用
  • {{ ... }} 表示模板替换
    • 类似 Python f-string
  • 可更动态，支持 for 循环

Jinja 模板示意图，使用双花括号进行替换

Python 中的 Apache Airflow 入门

Airflow 的 Jinja 函数

  • logical_date：Dag 运行发生的时间
  • ds：logical_date 的 YYYY-MM-DD 格式
  • logical_date.year、logical_date.month、logical_date.day：访问日期组件
  • params：Dag 运行时参数
  • 计算、循环等
    • macros.ds_add：支持日期计算
Python 中的 Apache Airflow 入门

使用 SmtpNotifier 的示例

  • 在主题字段中进行替换
@dag(dag_id=`sales_etl`,
     on_failure_callback=SmtpNotifier(
         to='[email protected]',
         from_email='[email protected]',

         subject='sales_etl on {{ logical_date }} has failed!'
     )
)
  • 主题：sales_etl 在 2026-04-30 失败！
Python 中的 Apache Airflow 入门

使用 FileSensor 的示例

  • 动态配置文件路径
  • 查找 ds 的前一天（YYYY-MM-DD 格式的 logical_date）的文件
  • 使用 macros.ds_add
    filewatcher = FileSensor(
task_id="wait_for_files",
filepath="/data/{{ macros.ds_add(ds, -1) }}/input.csv",
...
)
  • 若 ds 为 2026-04-30，路径为 /data/2026-04-29/input.csv
Python 中的 Apache Airflow 入门

Jinja 要点

  • 并非处处可用
  • 查看文档中标注为"可模板化"的属性
  • 检查代码视图

Airflow Dag 代码视图，显示定义 Dag 的只读 Python 源码

Python 中的 Apache Airflow 入门

开始练习吧！

Python 中的 Apache Airflow 入门

Preparing Video For Download...