Airflow 变量

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

什么是变量？

  • 适用于配置值
  • 如文件路径、API 端点、环境等
  • 可在任务中查询，通过代码、Airflow UI 或文件设置
  • 存在于 Dag 代码之外
Python 中的 Apache Airflow 入门

在 Airflow UI 中访问变量

  • 通过菜单 Admin: Variables

Airflow UI 打开 Admin 菜单并显示 Variables 选项

Python 中的 Apache Airflow 入门

在 Airflow UI 中查看变量

Airflow 变量页面，列表为空，未定义变量

Python 中的 Apache Airflow 入门

在 Airflow 中添加变量

Airflow 变量页面，高亮显示 Add Variable 按钮

Python 中的 Apache Airflow 入门

变量编辑器

Airflow 变量编辑器表单，包含 key、value 和 description 字段

Python 中的 Apache Airflow 入门

UI 中的变量

Airflow 变量页面，列表中显示一个已定义的变量

Python 中的 Apache Airflow 入门

在 Python 中读取变量

  • 使用 airflow.sdk.Variable 库
  • 通过 Variable.get("variable_name") 读取
    # Retrieve a plain string value
my_region = Variable.get("region")
  • 此时，my_region 为 us-eastern
  • 若变量不存在可提供回退值
    # Provide a fallback if the variable doesn't exist
env = Variable.get("current_environment", default=None)
  • 因未定义 current_environment，env 为 None
Python 中的 Apache Airflow 入门

在 Jinja 模板中访问变量

  • 可通过 {{ var.value.variable_name }} 访问
    filewatcher = FileSensor(
task_id="wait_for_files",
filepath="/data/{{ var.value.region }}/input.csv",
...
  • 在此例中，FileSensor 将查找文件：/data/us-eastern/input.csv
Python 中的 Apache Airflow 入门

通过 CLI 访问变量

  • 可通过命令行读取、更新、删除或列出变量
  • 读取 - airflow variables get <variable_name>
  • 写入 - airflow variables set <variable_name> <value>
  • 删除 - airflow variables delete <variable_name>
  • 列出 - airflow variables list <variable_name>
Python 中的 Apache Airflow 入门

JSON 变量

  • 变量值可为 JSON 字符串
    • Variable.get("variable_name", deserialize_json=True)
    • 返回 Python 字典
  • 在 Jinja 中
    • {{ var.json.variable_name }}
Python 中的 Apache Airflow 入门

设置变量

  • 也可以设置变量
  • 使用 Variable.set("variable_name", variable_value)
  • 也可通过 serialize_json=True 处理 JSON
  • 谨慎使用，否则可能以意外方式影响其他访问变量的 Dags
Python 中的 Apache Airflow 入门

¡Vamos a practicar!

Python 中的 Apache Airflow 入门

Preparing Video For Download...