Airflow工作流编排原理与Python DAG实战入门

1. 这不是又一个“Python框架”——Airflow到底在解决什么真问题？

Apache Airflow 不是 Python 的新语法糖，也不是让你多写几行print("Hello World")的玩具工具。它解决的是一个在数据工程、机器学习、ETL 和自动化运维中反复出现、却长期被野路子硬扛的痛点：当你的任务从“单步执行”变成“多步依赖、跨系统、需重试、要监控、得回溯”的复杂链条时，靠 shell 脚本 + cron + 邮件报警 + Excel 手动记日志，已经彻底崩盘了。我亲眼见过三个团队用不同方式“自研调度器”：一个用 Redis 做队列+Flask 写 API，一个用 MySQL 表存状态+定时轮询，还有一个直接把所有逻辑塞进一个超长的 Bash 脚本里，靠set -e和一堆if [ $? -ne 0 ]; then exit 1; fi硬撑。结果呢？上线三个月后，没人敢改脚本，没人能说清某次失败到底是上游数据没来，还是中间 Python 脚本内存溢出，还是下游数据库连接池满了——因为所有“为什么失败”和“现在卡在哪一步”的信息，都散落在三台服务器的日志文件里，靠人肉 grep 拼凑。

Airflow 的核心价值，就藏在它的名字里：Orchestration（编排），而不是 Scheduling（调度）。调度器只管“几点跑”，编排器管的是“谁先谁后、谁等谁、谁失败了怎么补、谁成功了通知谁、历史记录能不能查、权限能不能控”。它把整个工作流当成一个有向无环图（DAG），每个节点是一个可执行的原子任务（比如“从 S3 下载 CSV”、“用 Pandas 清洗数据”、“把结果写入 PostgreSQL”），边是任务之间的依赖关系（比如“清洗必须等下载完成之后才能开始”）。这个模型不是 Airflow 发明的，但它是第一个把 DAG 概念真正落地成开发者日常编码体验的开源项目——你不是在配置 XML 或 JSON，而是在写 Python 代码定义这个图。这意味着，你熟悉的def、if、for、异常处理、单元测试、Git 版本管理，全都能无缝迁移到工作流编排中。你写的不是“配置”，而是“可维护、可测试、可复用的业务逻辑”。所以，如果你正在为“每天凌晨 2 点跑的报表脚本突然不发邮件了，但日志里只有一行ERROR: NoneType”而头疼；或者你的机器学习 pipeline 每次 retrain 都要手动点五次按钮、检查四个地方的状态、再复制粘贴三次命令；又或者你的老板问“上个月 15 号那批订单数据，清洗环节到底卡在哪个步骤了？耗时多久？重试了几次？”，而你只能翻着终端历史记录含糊其辞——那么，Airflow 就不是“可学可不学”的新技术，而是你手头那把已经卷刃的瑞士军刀，该换一把带激光测距和指南针的专业工具了。它面向的不是“想学 Python 的小白”，而是“已经会写 Python 脚本，但正被脚本的规模和复杂度压得喘不过气”的真实从业者。

2. 核心设计哲学：为什么非得用 Python 写 DAG？这背后有三重深意

2.1 DAG 不是图，是“可执行的契约”

很多人第一次看到 Airflow 的 DAG 定义代码，第一反应是：“这不就是画个流程图吗？用 draw.io 不香吗？” 这是个根本性误解。Airflow 的 DAG 文件（.py）本质上是一份动态生成的、带有完整执行上下文的契约。它不只是描述“A 之后是 B”，而是声明：“B 的执行，必须严格满足 A 的返回值为True、且 A 的运行时长不超过 300 秒、且 A 的输出目录/data/raw/2024-05-20/必须存在且非空”。这个契约的每一个条款，都由 Python 代码精确表达。比如：

from airflow import DAG from airflow.operators.python import PythonOperator from airflow.sensors.filesystem import FileSensor from datetime import datetime, timedelta default_args = { 'owner': 'data-engineer', 'depends_on_past': False, 'start_date': datetime(2024, 5, 20), 'email_on_failure': True, 'retries': 3, 'retry_delay': timedelta(minutes=5) } dag = DAG( 'etl_daily_orders', default_args=default_args, description='Daily ETL for order data', schedule_interval='0 2 * * *', # 每天凌晨2点 catchup=False # 不补跑历史日期 ) # 传感器：等待上游系统生成的文件 wait_for_file = FileSensor( task_id='wait_for_raw_data', filepath='/data/upstream/orders_{{ ds }}.csv', # ds 是 Airflow 内置宏，代表当前执行日期 poke_interval=300, # 每5分钟检查一次 timeout=3600, # 最多等1小时，超时则任务失败 dag=dag ) # Python 任务：执行清洗逻辑 def clean_orders(): import pandas as pd # 读取昨天的文件 df = pd.read_csv(f'/data/upstream/orders_{datetime.now().date() - timedelta(days=1)}.csv') # 执行清洗 df = df.dropna(subset=['order_id']) df['amount'] = df['amount'].astype(float) df.to_csv(f'/data/cleaned/orders_cleaned_{datetime.now().date() - timedelta(days=1)}.csv', index=False) clean_task = PythonOperator( task_id='clean_order_data', python_callable=clean_orders, dag=dag ) # 依赖关系：clean_task 必须在 wait_for_file 成功后执行 wait_for_file >> clean_task

这段代码里，wait_for_file不是一个静态的“等待”动作，而是一个持续运行的传感器（Sensor），它会在后台不断轮询文件系统，直到条件满足或超时。clean_task也不是一个简单的函数调用，它被 Airflow 包装成一个独立的、可被重试、可被监控、可被单独触发的执行单元。>>符号定义的依赖，不是编译时的顺序，而是运行时的强约束：如果wait_for_file失败，clean_task绝对不会启动。这种将“业务规则”（等文件）、“执行逻辑”（清洗数据）、“运维策略”（重试3次、超时1小时）全部用同一门语言（Python）在同一份文件里声明的能力，是任何纯配置型工具无法比拟的。它让工作流的定义、开发、测试、部署，完全融入了现代软件工程的生命周期。

2.2 “Python 作为 DSL”：零成本降低认知门槛

Airflow 选择 Python 作为其领域特定语言（DSL），绝非偶然。对于一个主要用户是数据工程师、分析师、ML 工程师的项目来说，Python 是他们最熟悉、最没有学习负担的工具。想象一下，如果 Airflow 强制要求你用 YAML 写一个复杂的条件分支：

# 伪代码：YAML 版本的条件分支（实际 Airflow 不支持这么写） tasks: - name: check_data_quality operator: python python_callable: "lambda: check_quality()" on_success: - if: "{{ task_instance.xcom_pull(task_ids='check_data_quality') == 'good' }}" then: run_model_training - else: - send_alert - run_data_correction

这不仅冗长，而且失去了 Python 的所有优势：无法调试（你不能在 YAML 里加print()）、无法复用（不能import其他模块）、无法进行复杂的逻辑判断（YAML 的模板语法极其有限）。而用 Python，一切变得自然：

def decide_next_step(**context): quality_result = context['task_instance'].xcom_pull(task_ids='check_data_quality') if quality_result == 'good': context['task_instance'].xcom_push(key='next_task', value='run_model_training') else: context['task_instance'].xcom_push(key='next_task', value='run_data_correction') # 同时触发告警 send_slack_alert("Data quality issue detected!") decide_task = PythonOperator( task_id='decide_next_step', python_callable=decide_next_step, dag=dag ) # 使用 BranchPythonOperator 实现真正的分支 from airflow.operators.python import BranchPythonOperator branch_task = BranchPythonOperator( task_id='branch_on_quality', python_callable=lambda: 'run_model_training' if quality_result == 'good' else 'run_data_correction', dag=dag )

这里，BranchPythonOperator的python_callable是一个标准的 Python 函数，你可以像写任何其他 Python 代码一样，加断点、查文档、用 IDE 的自动补全。你不需要去学一套新的、小众的、只有 Airflow 才懂的“配置语法”。这种“零额外语法成本”的设计，是 Airflow 能在数据领域快速普及的关键。它不强迫你改变思维方式，而是把你已有的 Python 技能，直接平移过来解决更高级的问题。

2.3 DAG 文件即代码：版本控制、CI/CD 和协作的基石

当你的工作流定义是.py文件时，它就天然地成为了 Git 仓库里的一等公民。你可以给 DAG 文件写单元测试（用airflow.models.dag.DAG和airflow.utils.dates.days_ago模拟运行环境），可以在 CI 流水线里运行pytest tests/test_dags.py来确保新提交的 DAG 语法正确、依赖无误、关键路径可达。我曾经参与过一个项目，团队在合并 PR 前，CI 会自动执行以下检查：

1. python -m py_compile dags/etl_daily_orders.py—— 编译检查，确保没有语法错误；
2. airflow dags list-import-errors—— 检查 DAG 是否能被 Airflow 解析；
3. pytest tests/test_etl_daily_orders.py --cov=dags—— 运行单元测试，覆盖clean_orders函数的边界情况（如空文件、格式错误）；
4. airflow dags list-import-errors --output-json | jq '.dags[] | select(.dag_id=="etl_daily_orders")'—— 确保 DAG ID 在列表中。

这套流程，让“修改一个调度逻辑”这件事，从“登录到生产服务器，小心翼翼地编辑一个.py文件，祈祷别手抖”的高危操作，变成了“在本地写好代码，推送到 GitHub，等待 CI 绿灯，然后一键合并”的标准化交付。更重要的是，它解决了协作难题。当多个工程师共同维护一个复杂的金融风控 pipeline 时，A 工程师负责“特征计算”部分，B 工程师负责“模型评分”部分，C 工程师负责“结果推送”部分。他们各自在自己的分支上开发、测试自己的 DAG 片段，通过git merge和git diff，可以清晰地看到“这次合并引入了哪些新的任务、修改了哪些依赖、调整了哪些重试策略”。这种基于代码的协作模式，是任何图形化界面或配置中心都无法提供的透明度和可追溯性。

3. 从零搭建：一个可立即运行的本地 Airflow 环境（含避坑详解）

3.1 环境准备：为什么推荐pip而非conda或docker？

在开始安装前，必须明确一点：Airflow 的官方强烈推荐使用pip进行安装，尤其是在学习和开发阶段。这不是教条，而是有深刻的技术原因。Airflow 的核心是一个 Python 应用，但它严重依赖于大量 C 扩展库（如psycopg2用于 PostgreSQL、pymysql用于 MySQL、cryptography用于安全通信）。conda的包管理器虽然强大，但在处理这些需要编译的扩展时，常常会因为预编译二进制包的 ABI（应用二进制接口）不匹配而报错，典型错误如ImportError: libpq.so.5: cannot open shared object file。而 Docker 虽然能提供完美的隔离环境，但对于初学者来说，它引入了额外的认知负荷：你需要同时理解 Docker 的网络、卷挂载、镜像构建，以及 Airflow 的内部机制。当你第一次遇到webserver启动失败时，你无法确定问题是出在 Airflow 配置上，还是 Docker 的端口映射上，抑或是宿主机的防火墙设置上。

因此，我的建议是：在你的本地 Python 环境中，用pip安装一个精简版的 Airflow，专用于学习和实验。这样，所有的错误信息、日志、调试过程，都发生在你最熟悉的venv环境里，排查路径最短。

提示：请确保你的系统已安装 Python 3.8 或更高版本（Airflow 2.7+ 要求 Python 3.8+）。可以通过python --version和which python（macOS/Linux）或where python（Windows）确认。

3.2 分步安装与初始化：每一步背后的“为什么”

1. 创建并激活虚拟环境

   # 创建一个名为 airflow_env 的虚拟环境 python -m venv airflow_env # 激活它（macOS/Linux） source airflow_env/bin/activate # 激活它（Windows） airflow_env\Scripts\activate.bat

   注意：虚拟环境是 Python 项目的基石。它将 Airflow 及其所有依赖（可能多达 200+ 个包）与你系统全局的 Python 环境完全隔离开。这样，你升级 Airflow 不会影响你用pip install jupyter安装的 Jupyter Notebook，反之亦然。这是避免“Python 环境灾难”的唯一可靠方法。

2. 升级 pip 并安装 Airflow

   # 升级 pip 到最新版，避免旧版 pip 无法解析复杂的依赖树 pip install --upgrade pip # 安装 Airflow 核心，以及 SQLite 作为元数据库（学习用，足够了） pip install "apache-airflow[sqlite]" # 如果你后续想连接 PostgreSQL，再运行：pip install "apache-airflow[postgres]" # 如果你后续想连接 MySQL，再运行：pip install "apache-airflow[mysql]"

   关键解释：[sqlite]是一个“extras”标记。Airflow 的核心功能不依赖任何数据库，但它的元数据（DAG 定义、任务状态、日志、用户信息等）必须存储在一个数据库中。SQLite 是一个轻量级的、单文件的嵌入式数据库，它不需要单独安装服务、配置用户密码、管理连接池。对于学习和本地开发，它是完美的选择。它把所有元数据都存在你本地磁盘上的一个airflow.db文件里，简单、直观、零运维。而postgres或mysql这些“生产级”数据库，则需要你额外安装、配置、维护一个数据库服务，这在入门阶段完全是不必要的负担。

3. 初始化数据库

   # 初始化 Airflow 的元数据库（会创建 airflow.db 文件） airflow db init

   这一步会根据你当前的 Airflow 配置（默认是 SQLite），在当前目录下创建airflow.db文件，并在里面建立所有必需的数据表（如dag,task_instance,log,user等）。这是 Airflow 能够“记住”你定义了哪些 DAG、哪些任务成功了、哪些失败了的基础。如果这一步报错，90% 的原因是你的 Python 环境没有正确激活，或者pip install没有成功。

4. 创建初始用户

   # 创建一个管理员用户，用户名和密码都是 airflow airflow users create \ --username airflow \ --password airflow \ --firstname Peter \ --lastname Parker \ --role Admin \ --email peter@parkercorp.com

   Airflow 的 Web UI 是一个完整的 Web 应用，它有自己的用户认证和权限系统。Admin角色拥有最高权限，可以查看所有 DAG、触发所有任务、管理所有用户。记住这个用户名和密码，稍后登录 Web UI 就靠它了。

5. 启动 Web Server 和 Scheduler

   # 在一个终端窗口中，启动 Web Server（监听 8080 端口） airflow webserver # 在另一个终端窗口中，启动 Scheduler（它负责解析 DAG、触发任务、更新状态） airflow scheduler

   这是 Airflow 的两个核心进程。webserver是你和 Airflow 交互的“前台”，它提供 Web UI 和 REST API。scheduler是默默工作的“后台”，它就像一个不知疲倦的交通警察，时刻盯着所有 DAG 的schedule_interval，一旦时间到了，就立刻生成一个DagRun（一次 DAG 的执行实例），然后按依赖关系，依次触发其中的TaskInstance（任务实例）。这两个进程必须同时运行，Airflow 才能正常工作。如果你只启动了webserver，你会看到 UI，但所有 DAG 都是灰色的，因为没有scheduler去告诉它们“该跑了”。

3.3 验证与第一个 DAG：让代码真正“动起来”

1. 打开浏览器，访问http://localhost:8080。输入用户名airflow和密码airflow登录。
2. 找到DAGs菜单。此时，你应该能看到一个名为example_python_operator的 DAG。这是 Airflow 自带的示例，它展示了如何用 PythonOperator 执行一个简单的函数。点击它，进入 DAG 的详情页。
3. 点击右上角的Trigger DAG按钮。这会手动触发一次该 DAG 的执行。
4. 回到DAGs列表页，刷新页面。你会发现example_python_operator的状态从灰色变成了绿色（表示正在运行），过几秒钟，它会变成蓝色（表示成功）。
5. 点击 DAG 名称，进入 Graph View。你会看到一个清晰的流程图，显示了print_the_context和print_dag_run_conf两个任务，以及它们之间的箭头。点击任何一个任务节点，可以查看它的详细日志（Log），里面会打印出Hello World和一些上下文信息。

实操心得：第一次看到 DAG 在 UI 上从灰色变绿，再变蓝，那种“我写的代码真的在被一个强大的系统调度执行”的感觉，是学习 Airflow 最大的动力来源。不要跳过这一步，一定要亲手触发、亲眼看到、亲耳听到（日志里的INFO信息）。

4. 核心概念与实操：DAG、Operator、Task、XCom 如何协同工作？

4.1 DAG：工作流的蓝图，而非执行体

一个 DAG（Directed Acyclic Graph）对象，在 Airflow 中，它本身并不执行任何东西，它只是一个定义、一个蓝图、一个模板。它定义了“有哪些任务”、“它们之间是什么依赖关系”、“这个工作流多久运行一次”、“失败了怎么办”等元信息。真正干活的是 DAG 的“实例”，即DagRun。

• DAG 对象：由你的.py文件定义，存在于 Airflow 的 Python 进程内存中。它被scheduler定期扫描、解析。
• DagRun 对象：当scheduler根据schedule_interval（例如0 2 * * *）判断到某个时间点该运行这个 DAG 时，它就会创建一个DagRun实例。这个实例会被持久化到元数据库（airflow.db）中，记录下这次运行的execution_date（执行日期）、state（状态：running/success/failed）等。
• TaskInstance 对象：DagRun创建后，scheduler会根据 DAG 中定义的依赖关系，依次创建TaskInstance。每一个TaskInstance对应 DAG 中的一个具体任务（如wait_for_file）在某一次DagRun中的具体执行。它的状态（queued/running/success/failed/upstream_failed）也会被记录在数据库中。

这个三层结构（DAG -> DagRun -> TaskInstance）是理解 Airflow 运行时模型的关键。它解释了为什么你可以“暂停”一个 DAG（只是不让scheduler创建新的DagRun），但已经创建的DagRun会继续执行；也解释了为什么你可以“清除”一个 DAG 的历史任务（删除TaskInstance记录），但 DAG 的定义（蓝图）依然完好无损。

4.2 Operator：任务的“类型”与“行为”封装

Operator 是 Airflow 的“积木”。它定义了“一个任务应该做什么”。Airflow 提供了上百种内置 Operator，覆盖了绝大多数场景：

• PythonOperator：执行一个 Python 函数。这是最常用、最灵活的 Operator，适合任何可以用 Python 表达的逻辑。
• BashOperator：执行一条 Bash 命令。适合调用 shell 脚本、curl请求、grep日志等。
• PostgresOperator：执行一条 SQL 语句到 PostgreSQL 数据库。
• S3ListOperator：列出 S3 存储桶中的文件。
• FileSensor：等待一个文件或目录出现（前面例子中用过）。
• EmailOperator：发送一封邮件。

选择哪个 Operator，取决于你的任务本质。原则是：优先选择最“窄”的 Operator。例如，如果你的任务只是“往 PostgreSQL 里插入一行数据”，那么PostgresOperator就比PythonOperator更合适，因为它更专注、更安全、更易审计。PythonOperator是万能的，但也是最“宽”的，它把所有责任（连接数据库、处理异常、管理事务）都交给了你写的函数，增加了出错的可能性。

注意：Operator 本身不包含任何业务逻辑。PythonOperator的python_callable参数才是你的业务逻辑。Operator 只是负责“在正确的时机，以正确的方式，调用你的业务逻辑”。

4.3 Task：Operator 的“实例化”，工作流的最小执行单元

当你在 DAG 文件中写下：

clean_task = PythonOperator( task_id='clean_order_data', python_callable=clean_orders, dag=dag )

你创建的不是一个“任务”，而是一个Task对象。这个Task对象是PythonOperator类的一个实例，它被绑定到了dag这个 DAG 对象上。它包含了所有关于这个任务的“静态”信息：task_id（唯一标识）、python_callable（要执行的函数）、retries（重试次数）、timeout（超时时间）等。

这个Task对象，就是工作流图中那个圆圈（Node）。而TaskInstance，则是这个圆圈在某一次具体执行时的“化身”。你可以把Task理解为“类（Class）”，把TaskInstance理解为“对象（Object）”。

4.4 XCom：任务间传递“小数据”的秘密通道

在复杂的 DAG 中，一个任务的输出，往往是下一个任务的输入。例如，“下载任务”下载了一个 CSV 文件，生成了文件路径/data/raw/orders_2024-05-20.csv；“清洗任务”需要读取这个路径。你不能把路径硬编码在清洗任务里，因为每天的日期都不同。Airflow 提供了XCom（Cross-Communication）机制来解决这个问题。

XCom 是一个轻量级的、键值对（key-value）的存储，它与TaskInstance绑定。一个任务可以push（推送）一个值到 XCom，另一个任务可以pull（拉取）它。

def download_data(**context): # 模拟下载，生成一个文件路径 file_path = f"/data/raw/orders_{context['ds']}.csv" # 将 file_path 推送到 XCom，key 默认为 'return_value' return file_path # PythonOperator 会自动将返回值 push 到 XCom def clean_data(**context): # 从上一个任务（download_data）的 XCom 中拉取 'return_value' file_path = context['task_instance'].xcom_pull(task_ids='download_data') print(f"Cleaning file: {file_path}") # ... 执行清洗逻辑 download_task = PythonOperator( task_id='download_data', python_callable=download_data, dag=dag ) clean_task = PythonOperator( task_id='clean_data', python_callable=clean_data, dag=dag ) download_task >> clean_task

提示：XCom 不是用来传大文件的！它的设计初衷是传递“小数据”，比如一个文件路径、一个 API 返回的 JSON 中的某个 ID、一个布尔标志位。Airflow 默认的 SQLite 后端对 XCom 的大小有限制（通常 48KB），如果试图推送一个 10MB 的 CSV 文件内容，会直接失败。对于大数据传输，请使用外部存储（S3、HDFS）作为中介，XCom 只传递指向它的“指针”。

5. 常见问题与排查技巧实录：那些让我熬夜到凌晨三点的坑

5.1 问题速查表：高频故障与一招解决

5.2 独家避坑技巧：来自血泪教训的 3 条铁律

铁律一：永远不要在default_args里设置start_date为datetime.now()

这是一个新手几乎 100% 会踩的坑。start_date不是“这个 DAG 什么时候开始生效”，而是“这个 DAG 的第一次DagRun的execution_date是什么”。Airflow 的调度逻辑是：DagRun的execution_date是schedule_interval的上一个周期的开始时间。例如，schedule_interval='0 2 * * *'（每天凌晨2点），那么execution_date就是前一天的00:00:00。如果你把start_date设为datetime.now()，那么scheduler会认为“从现在开始，才允许创建DagRun”，但它会立刻尝试创建一个execution_date为datetime.now()的DagRun，而这与schedule_interval的语义是冲突的，导致 DAG 无法被正确调度。

正确做法：start_date应该是一个固定的、过去的日期时间，比如datetime(2024, 1, 1)。它标志着“从这一天起，这个 DAG 开始参与调度”。Airflow 会自动为你补跑从start_date到今天的每一个符合schedule_interval的DagRun（除非你设置了catchup=False）。

铁律二：catchup=False不是“关闭补跑”，而是“关闭历史补跑”

很多教程说“设置catchup=False可以避免 Airflow 启动时疯狂创建几百个DagRun”。这没错，但它的真实含义是：“当这个 DAG 第一次被scheduler发现时，不要为start_date到now()之间的所有周期都创建DagRun，只创建now()之后的第一个周期。” 这对于一个全新的、只关心未来数据的 DAG 是完美的。但如果你有一个修复 bug 的 DAG，需要重新处理过去一周的数据，catchup=False就会成为障碍。这时，你应该：

1. 临时将catchup=True；
2. 在 UI 中，找到该 DAG，点击Trigger DAG旁边的...，选择Clear，勾选Past和Future，然后Confirm。这会为指定的时间范围（比如2024-05-15到2024-05-21）创建新的DagRun；
3. 处理完后，再把catchup改回False。

铁律三：PythonOperator的函数，永远不要有副作用（Side Effect）

副作用是指函数除了返回值之外，还修改了外部状态（如全局变量、文件系统、数据库）。PythonOperator的python_callable函数，可能会被 Airflow 多次调用（比如重试时），也可能在不同的进程中被调用（LocalExecutor下）。如果你的函数里写了with open('log.txt', 'a') as f: f.write('ran!')，那么每次重试，都会往log.txt里追加一行，导致日志混乱。更危险的是，如果你的函数里写了db.execute("UPDATE table SET status='processed' WHERE id=123")，那么重试就会导致数据库被更新多次。

正确做法：将所有“有状态”的操作，都封装在幂等（Idempotent）的逻辑里。幂等的意思是：“无论执行一次还是执行一百次，最终结果都是一样的”。例如，更新数据库时，不要用UPDATE ... SET status='processed'，而是用UPDATE ... SET status='processed' WHERE status != 'processed'。或者，更好的方式是，把“状态”作为任务的输入（通过 XCom 或context），让任务的执行结果（成功/失败）本身，就成为状态的唯一权威来源。

6. 从入门到进阶：你的第一个生产级 DAG 应该如何设计？

6.1 场景还原：一个真实的电商数据同步需求

假设你是一家电商公司的数据工程师。每天凌晨 1 点，上游的订单系统会将前一天的订单数据，以 CSV 格式，上传到公司内网的 SFTP 服务器上。你的任务是：

1. 从 SFTP 下载这个 CSV 文件；
2. 用 Pandas 进行数据清洗（去重、填充缺失值、类型转换）；
3. 将清洗后的数据，加载到公司内部的 PostgreSQL 数据仓库中；
4. 如果任何一步失败，发送 Slack 告警给数据团队；
5. 如果成功，发送一封汇总邮件给业务部门，告知“昨日订单数据已就绪”。

这个需求，完美涵盖了 Airflow 的核心能力：外部系统集成（SFTP）、数据处理（Python）、数据库操作（PostgreSQL）、通知（Slack/Email）、错误处理（重试、告警）。

6.2 架构设计：一个健壮 DAG 的 5 个关键层

一个生产级的 DAG，不应该是一个扁平的、线性的任务链。它应该分层设计，每一层承担明确的职责：

1. 接入层（Ingestion Layer）：负责与外部世界“握手”。任务如sftp_sensor（等待 SFTP 上的文件出现）、sftp_download（下载文件）。这一层的特点是：高 IO、低 CPU、易失败（网络波动）。因此，要配置较长的poke_interval（传感器轮询间隔）和较多的retries。
2. 处理层（Processing Layer）：负责核心的业务逻辑。任务如clean_data（Pandas 清洗）、validate_data（数据质量校验，如检查订单金额是否为负数）。这一层的特点是