4. Airflow 实战：编写第一个 DAG、任务依赖设置、调度周期配置、变量与连接管理

好，咱们开始动手。前面聊了那么多理论，现在终于要写代码了。Airflow 的核心就是 DAG——说白了，就是一张有向无环图，把一个个任务串起来。这一章我会带你从零写一个 DAG，然后一步步加上依赖、调度、变量和连接。嗯，都是我在生产环境里天天用的东西。

4.1 第一个 DAG：Hello World 的升级版

先别急着搞复杂的东西。我们写一个最简单的 DAG，让它每天打印一句 "Hello from Airflow"。你想想看，这就像学编程时的 "Hello World"，但咱们这个能真正跑在调度系统里。

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator

def print_hello():
    print("Hello from Airflow! 当前时间: ", datetime.now())

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'hello_airflow_dag',
    default_args=default_args,
    description='我的第一个 DAG',
    schedule_interval='0 8 * * *',  # 每天早上8点
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['example'],
)

task_hello = PythonOperator(
    task_id='say_hello',
    python_callable=print_hello,
    dag=dag,
)

这段代码放到 $AIRFLOW_HOME/dags/ 目录下，Airflow 会自动解析。你可以在 Web UI 的 DAGs 页面看到它。嗯，这里要注意：文件名就是 DAG ID 的一部分，别乱起名。

4.2 任务依赖设置：让任务按顺序跑

实际作业流不可能只有一个任务。比如我们做数据 pipeline，通常是：抽取 → 转换 → 加载。这三个步骤有先后顺序。Airflow 里设置依赖非常简单，用位移运算符就行。

from airflow.operators.bash import BashOperator

# 定义三个任务
task_extract = BashOperator(
    task_id='extract_data',
    bash_command='echo "抽取数据..."',
    dag=dag,
)

task_transform = BashOperator(
    task_id='transform_data',
    bash_command='echo "转换数据..."',
    dag=dag,
)

task_load = BashOperator(
    task_id='load_data',
    bash_command='echo "加载数据..."',
    dag=dag,
)

# 设置依赖：抽取 → 转换 → 加载
task_extract >> task_transform >> task_load

这行 task_extract >> task_transform >> task_load 就是依赖关系。箭头指向谁，谁就等前面的跑完。你也可以用 set_downstream() 方法，但我个人觉得位移运算符更直观。

我在项目中遇到过一种坑：任务依赖写成了循环依赖。比如 A→B→C→A，Airflow 会直接报错，DAG 都解析不出来。所以画 DAG 图的时候，脑子里过一遍有没有环。

4.3 调度周期配置：cron 表达式与 timedelta

调度周期是 DAG 的灵魂。Airflow 支持两种方式：cron 表达式和 timedelta 对象。我一般用 cron，因为更精确。

如果你用 timedelta，写法是这样的：

dag = DAG(
    'my_timedelta_dag',
    schedule_interval=timedelta(hours=1),  # 每小时跑一次
    ...
)

但 timedelta 有个问题：它没有「对齐到整点」的概念。比如你 8:23 启动 DAG，下一次就是 9:23，再下一次 10:23。而 cron 表达式会严格对齐到整点。我个人建议：生产环境用 cron，开发测试可以用 timedelta 图省事。

4.4 变量与连接管理：别再硬编码了

你想想看，如果数据库连接串、 这些东西直接写在 DAG 文件里，那每次换环境都得改代码。Airflow 提供了两个好东西：Variables（变量）和 Connections（连接）。

4.4.1 Variables：存配置参数

Variables 就是键值对。你可以在 Web UI 的 Admin → Variables 里添加，也可以用 CLI 或代码操作。

from airflow.models import Variable

# 获取变量
db_name = Variable.get("db_name", default_var="default_db")
# 设置变量
Variable.set("api_endpoint", "https://api.example.com/v1")
# 删除变量
Variable.delete("temp_config")

4.4.2 Connections：管理外部系统连接

Connections 比 Variables 更强大，它专门用来存数据库、SSH、HTTP 等外部系统的连接信息。你可以在 Web UI 的 Admin → Connections 里配置。

在 DAG 里使用连接：

from airflow.hooks.base import BaseHook

# 获取连接对象
conn = BaseHook.get_connection("my_mysql_conn")
# 拿到具体信息
host = conn.host
port = conn.port
login = conn.login
password = conn.password
schema = conn.schema  # 数据库名

我在项目中遇到过最头疼的事：测试环境和生产环境的连接信息不一样，但 DAG 代码是同一份。解决方案就是连接名保持一致，比如都叫 prod_mysql，然后在不同环境的 Airflow 里配置不同的连接信息。代码不用改，只改配置。

4.5 本章知识体系总览

下面这张图把这一章的核心逻辑串起来了。你可以看到 DAG 从定义、任务编排、调度配置到外部依赖管理，是一条完整的链路。

这一章的内容就到这里。你照着上面的代码写一个 DAG，放到 Airflow 里跑起来，就能看到任务在 Web UI 上跳动了。记住：先跑通最简单的，再慢慢加依赖和配置。别一上来就想搞复杂的 pipeline，容易把自己绕进去。