未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

数据管道实战

---

案例背景

本案例为教学假设，并非已核实的企业项目；每天约 10 万新增订单及以下质量阈值均为示例设定。我们要为一个假设的电商平台构建每日数据管道，需求如下：

• 数据源：MySQL 业务库（orders 表，每天约 10 万新增订单）
• 处理流程：抽取 → 质量检查 → 清洗 → 写入 Hive ODS → 聚合到 DWS → 生成报表 → 邮件通知
• 调度要求：每天凌晨 2 点运行，处理前一天的数据
• 质量要求：
  • 数据量不为零
  • 金额字段空值率 < 5%
  • 订单 ID 无重复
• 故障处理：失败时发钉钉告警，自动重试 2 次

---

项目结构

dags/
├── ecommerce_daily_etl.py      # 主 DAG 文件
└── utils/
    ├── __init__.py
    ├── mysql_extractor.py      # MySQL 数据抽取逻辑
    ├── data_cleaner.py         # 数据清洗逻辑
    ├── quality_checker.py      # 数据质量检查
    ├── hive_loader.py          # Hive 写入逻辑
    └── alerting.py             # 告警通知

---

完整 DAG 代码

# dags/ecommerce_daily_etl.py
"""
电商平台每日 ETL 数据管道
数据流：MySQL → ODS（Hive）→ DWS（Hive）→ MySQL（报表）→ 邮件通知

架构设计：
- 数据写入步骤采用覆盖或替换思路；事务故障恢复、ETL 时间字段与邮件去重仍需补齐验证，不能保证每个步骤严格幂等
- 通过 XCom 传递轻量级元数据（行数、文件路径）
- 本示例通过本地文件路径传递数据（不通过 XCom）；跨 worker 或 Hive 执行环境运行须配置共享存储或补充文件传输
"""

from datetime import datetime, timedelta
import os

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.providers.apache.hive.operators.hive import HiveOperator
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.operators.email import EmailOperator
from airflow.utils.task_group import TaskGroup

# ============================================================
# 工具函数
# ============================================================

def send_alert(context):
    """钉钉告警（失败回调）"""
    import requests
    import json

    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    execution_date = str(context["ds"])
    log_url = task_instance.log_url

    webhook_url = os.environ.get("DINGTALK_WEBHOOK_URL", "")
    if not webhook_url:
        print("[WARN] DINGTALK_WEBHOOK_URL 未配置，跳过告警")
        return

    msg = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"Airflow 告警：{dag_id}",
            "text": (
                f"## Airflow ETL 失败\n\n"
                f"- **DAG**: {dag_id}\n"
                f"- **Task**: {task_id}\n"
                f"- **日期**: {execution_date}\n"
                f"- [查看日志]({log_url})"
            ),
        },
    }
    try:
        requests.post(webhook_url, json=msg, timeout=10)
    except Exception as e:
        print(f"告警发送失败：{e}")


# ============================================================
# DAG 定义
# ============================================================

with DAG(
    dag_id="ecommerce_daily_etl",
    description="电商每日 ETL：MySQL → ODS → DWS → 报表 → 邮件",
    schedule_interval="0 2 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,           # 避免并发补数冲垮系统
    tags=["电商", "ETL", "每日"],
    default_args={
        "owner": "data_engineering",
        "retries": 2,
        "retry_delay": timedelta(minutes=10),
        "retry_exponential_backoff": True,
        "on_failure_callback": send_alert,
        "email_on_failure": False,   # 使用自定义告警，关闭邮件
        "execution_timeout": timedelta(hours=3),
    },
) as dag:

    # ============================================================
    # Task 1：抽取 MySQL 数据到本地 CSV（按日期增量）
    # ============================================================
    def extract_mysql_incremental(**context):
        """
        从 MySQL 按逻辑日期抽取订单数据到本地 CSV
        重跑设计：覆盖同一路径的 CSV；这里未实现 HDFS 上传或目录删除
        """
        import pandas as pd
        from sqlalchemy import create_engine
        from airflow.hooks.base import BaseHook

        ds = context["ds"]  # 逻辑日期；是否对应前一天须结合调度区间与时区确认
        print(f"开始抽取 {ds} 的订单数据")

        # 获取 MySQL 连接
        conn = BaseHook.get_connection("mysql_ecommerce_prod")
        engine = create_engine(
            f"mysql+pymysql://{conn.login}:{conn.password}"
            f"@{conn.host}:{conn.port}/{conn.schema}"
        )

        # 增量查询（只取前一天的数据）
        sql = f"""
            SELECT
                order_id,
                user_id,
                product_id,
                amount,
                status,
                city,
                channel,
                create_time,
                update_time,
                DATE(create_time) AS partition_date
            FROM orders
            WHERE DATE(create_time) = '{ds}'
        """
        df = pd.read_sql(sql, engine)
        row_count = len(df)
        print(f"查询到 {row_count} 行数据")

        # 写入临时 CSV（后续由 Hive LOAD DATA 导入）
        output_path = f"/data/staging/orders/dt={ds}/orders.csv"
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        df.to_csv(output_path, index=False)

        print(f"数据已写入：{output_path}")

        # 通过 XCom 传递元数据（不传实际数据）
        return {
            "row_count": row_count,
            "output_path": output_path,
            "ds": ds,
        }

    extract_task = PythonOperator(
        task_id="extract_mysql",
        python_callable=extract_mysql_incremental,
    )

    # ============================================================
    # Task 2：数据质量检查
    # ============================================================
    def run_quality_checks(**context):
        """
        数据质量检查，任何一项不通过就抛出异常，让 Task 失败
        """
        import pandas as pd

        ti = context["ti"]
        ds = context["ds"]

        # 从上游拿到文件路径和行数
        meta = ti.xcom_pull(task_ids="extract_mysql")
        output_path = meta["output_path"]
        expected_count = meta["row_count"]

        # 读取数据
        df = pd.read_csv(output_path)
        actual_count = len(df)

        print(f"开始数据质量检查：{actual_count} 行")

        # 检查 1：数据量不为零
        assert actual_count > 0, f"数据量为零！日期：{ds}"

        # 检查 2：数据量一致性（CSV 行数应与查询行数一致）
        assert actual_count == expected_count, (
            f"数据量不一致：查询 {expected_count} 行，文件 {actual_count} 行"
        )

        # 检查 3：订单 ID 唯一性
        dup_rate = df.duplicated(subset=["order_id"]).mean()
        assert dup_rate == 0, f"存在重复 order_id，重复率：{dup_rate:.2%}"

        # 检查 4：金额空值率
        null_rate_amount = df["amount"].isnull().mean()
        assert null_rate_amount < 0.05, (
            f"amount 空值率 {null_rate_amount:.2%} 超过 5% 阈值"
        )

        # 检查 5：金额合理性（不能有负数）
        negative_count = (df["amount"] < 0).sum()
        assert negative_count == 0, f"存在 {negative_count} 条金额为负的记录"

        print("所有质量检查通过！")
        print(f"  - 数据量：{actual_count} 行")
        print(f"  - 重复率：{df.duplicated(subset=['order_id']).mean():.4f}")
        print(f"  - amount 空值率：{null_rate_amount:.4f}")

        return {"quality_passed": True, "row_count": actual_count}

    quality_check = PythonOperator(
        task_id="quality_check",
        python_callable=run_quality_checks,
        # 质量检查失败不重试（失败是有意义的信号）
        retries=0,
    )

    # ============================================================
    # Task 3：数据清洗
    # ============================================================
    def clean_data(**context):
        """
        数据清洗：
        1. 过滤状态异常的订单（只保留有效状态）
        2. 标准化 city 字段
        3. 填充缺失值
        """
        import pandas as pd

        ti = context["ti"]
        ds = context["ds"]

        meta = ti.xcom_pull(task_ids="extract_mysql")
        input_path = meta["output_path"]
        output_path = input_path.replace("/staging/", "/cleaned/")

        df = pd.read_csv(input_path)
        original_count = len(df)

        # 清洗操作
        valid_statuses = {"paid", "cancelled", "refunded", "pending"}
        df["status"] = df["status"].str.strip().str.lower()
        df = df[df["status"].isin(valid_statuses)]

        df["city"] = df["city"].fillna("unknown").str.strip()
        df["channel"] = df["channel"].fillna("unknown").str.strip().str.lower()

        # 金额精度处理
        df["amount"] = df["amount"].round(2)

        cleaned_count = len(df)
        print(f"清洗完成：{original_count} → {cleaned_count} 行（过滤 {original_count - cleaned_count} 行）")

        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        df.to_csv(output_path, index=False)

        return {"cleaned_count": cleaned_count, "cleaned_path": output_path}

    clean_task = PythonOperator(
        task_id="clean_data",
        python_callable=clean_data,
    )

    # ============================================================
    # Task 4：加载到 Hive ODS 层
    # ============================================================
    load_to_ods = HiveOperator(
        task_id="load_to_ods",
        hql="""
            -- 先删除当日分区（幂等设计）
            ALTER TABLE ods.orders DROP IF EXISTS PARTITION (dt='{{ ds }}');

            -- 从外部文件加载数据
            LOAD DATA LOCAL INPATH '/data/cleaned/orders/dt={{ ds }}/orders.csv'
            INTO TABLE ods.orders
            PARTITION (dt='{{ ds }}');

            -- 验证加载行数
            SELECT COUNT(*) FROM ods.orders WHERE dt = '{{ ds }}';
        """,
        hive_cli_conn_id="hive_default",
    )

    # ============================================================
    # Task 5：聚合到 DWS 层（销售汇总）
    # ============================================================
    aggregate_to_dws = HiveOperator(
        task_id="aggregate_to_dws",
        hql="""
            INSERT OVERWRITE TABLE dws.order_daily_summary
            PARTITION (dt='{{ ds }}')
            SELECT
                city,
                channel,
                COUNT(order_id)                            AS order_count,
                COUNT(DISTINCT user_id)                   AS unique_user_count,
                SUM(amount)                               AS total_amount,
                AVG(amount)                               AS avg_amount,
                SUM(IF(status='cancelled', 1, 0))         AS cancel_count,
                SUM(IF(status='cancelled', 1, 0)) /
                    COUNT(order_id)                       AS cancel_rate,
                CURRENT_TIMESTAMP()                       AS etl_time
            FROM ods.orders
            WHERE dt = '{{ ds }}'
            GROUP BY city, channel
        """,
        hive_cli_conn_id="hive_default",
    )

    # ============================================================
    # Task 6：将报表数据同步到 MySQL（供 BI 工具使用）
    # ============================================================
    sync_to_mysql = SQLExecuteQueryOperator(
        task_id="sync_report_to_mysql",
        conn_id="mysql_bi_report",
        sql="""
            -- 先删除当日数据（幂等）
            DELETE FROM report_order_daily WHERE report_date = '{{ ds }}';

            -- 插入新数据（实际生产中应通过 Spark 或 sqoop 做大批量同步）
            INSERT INTO report_order_daily (report_date, city, channel,
                order_count, unique_user_count, total_amount, avg_amount,
                cancel_rate, create_time)
            SELECT
                '{{ ds }}' AS report_date,
                city, channel, order_count, unique_user_count,
                total_amount, avg_amount, cancel_rate,
                NOW() AS create_time
            FROM dws_order_daily_summary_external  -- 通过外部表读取 Hive
            WHERE dt = '{{ ds }}'
        """,
        autocommit=True,
    )

    # ============================================================
    # Task 7：生成汇总报告，发送邮件
    # ============================================================
    def generate_summary_report(**context):
        """生成 HTML 格式的日报汇总"""
        from airflow.hooks.base import BaseHook
        from sqlalchemy import create_engine
        import pandas as pd

        ds = context["ds"]

        # 从 BI 数据库读取今日汇总
        conn = BaseHook.get_connection("mysql_bi_report")
        engine = create_engine(
            f"mysql+pymysql://{conn.login}:{conn.password}"
            f"@{conn.host}:{conn.port}/{conn.schema}"
        )

        df = pd.read_sql(
            f"SELECT * FROM report_order_daily WHERE report_date = '{ds}' "
            f"ORDER BY total_amount DESC",
            engine
        )

        total_orders = df["order_count"].sum()
        total_amount = df["total_amount"].sum()

        # 生成 HTML 报表
        html_table = df.head(10).to_html(
            index=False,
            float_format=lambda x: f"{x:,.2f}",
        )

        html_content = f"""
        <html><body>
        <h2>电商平台每日销售日报 - {ds}</h2>
        <hr>
        <h3>核心指标</h3>
        <ul>
            <li>总订单数：<b>{total_orders:,}</b> 单</li>
            <li>总订单金额（含 paid、cancelled、refunded、pending 状态，非实收销售额）：<b>¥ {total_amount:,.2f}</b></li>
            <li>单均订单金额（上述订单金额合计 / 订单数）：<b>¥ {total_amount / total_orders if total_orders > 0 else 0:,.2f}</b></li>
        </ul>
        <h3>Top 10 城市/渠道汇总</h3>
        {html_table}
        <hr>
        <p style="color:gray; font-size:12px;">
            此邮件由 Airflow 自动发送 | DAG: ecommerce_daily_etl | 日期: {ds}
        </p>
        </body></html>
        """

        return html_content

    generate_report = PythonOperator(
        task_id="generate_report",
        python_callable=generate_summary_report,
    )

    send_email = EmailOperator(
        task_id="send_report_email",
        to=["manager@company.com", "data-team@company.com"],
        subject="电商每日销售日报 - {{ ds }}",
        html_content="{{ task_instance.xcom_pull(task_ids='generate_report') }}",
    )

    # ============================================================
    # 定义任务依赖关系
    # ============================================================
    (
        extract_task
        >> quality_check
        >> clean_task
        >> load_to_ods
        >> aggregate_to_dws
        >> sync_to_mysql
        >> generate_report
        >> send_email
    )

---

增量处理模式

上面的案例是最简单的增量模式（按日期增量）。实际场景中有几种常见的增量模式：

模式一：按分区日期增量（推荐，适合大多数场景）

def extract_by_partition(**context):
    """按 partition_date 字段增量抽取"""
    ds = context["ds"]

    sql = f"""
        SELECT *
        FROM orders
        WHERE DATE(create_time) = '{ds}'
    """
    # 查询并写入 HDFS

特点：

• 简单可靠
• 需要源数据有清晰的时间字段
• 每次处理固定时间范围的数据

模式二：按自增 ID 增量（适合 ID 严格递增的表）

def extract_by_id(**context):
    """按自增 ID 增量抽取，通过 Airflow Variable 记录 watermark"""
    from airflow.models import Variable

    ds = context["ds"]

    # 读取上次同步的最大 ID
    last_id = int(Variable.get("orders_last_sync_id", default_var=0))

    sql = f"""
        SELECT *
        FROM orders
        WHERE id > {last_id}
        ORDER BY id
        LIMIT 100000  -- 每次最多同步 10 万条，防止超时
    """

    df = execute_query(sql)

    if len(df) > 0:
        # 更新 watermark
        new_last_id = df["id"].max()
        Variable.set("orders_last_sync_id", str(new_last_id))
        print(f"同步完成：{last_id} → {new_last_id}，共 {len(df)} 行")
    else:
        print(f"无新增数据，当前 ID：{last_id}")

    return {"synced_count": len(df)}

模式三：Change Data Capture（CDC）模式

CDC 是最彻底的增量方案，直接捕获数据库的增删改操作。通常结合 Flink CDC 或 Debezium 实现，Airflow 负责调度和编排。

def trigger_flink_cdc_sync(**context):
    """触发 Flink CDC 同步作业"""
    # 调用 Flink REST API 触发指定时间范围的 CDC 同步
    # 这里只是示意，实际需要根据 Flink 作业设计来实现
    pass

---

数据质量检查集成

更完整的数据质量框架设计：

class DataQualityChecker:
    """可复用的数据质量检查类"""

    def __init__(self, df, table_name: str, ds: str):
        self.df = df
        self.table_name = table_name
        self.ds = ds
        self.results = []

    def check_not_empty(self):
        """检查数据量不为零"""
        count = len(self.df)
        passed = count > 0
        self.results.append({
            "check": "not_empty",
            "passed": passed,
            "value": count,
            "threshold": ">0",
            "message": f"数据量：{count}" if passed else f"数据量为零！"
        })
        return self

    def check_null_rate(self, column: str, threshold: float = 0.05):
        """检查指定列的空值率"""
        null_rate = self.df[column].isnull().mean()
        passed = null_rate < threshold
        self.results.append({
            "check": f"null_rate_{column}",
            "passed": passed,
            "value": null_rate,
            "threshold": f"<{threshold}",
            "message": f"{column} 空值率：{null_rate:.2%}"
        })
        return self

    def check_uniqueness(self, columns: list):
        """检查唯一性"""
        dup_rate = self.df.duplicated(subset=columns).mean()
        passed = dup_rate == 0
        self.results.append({
            "check": f"uniqueness_{'_'.join(columns)}",
            "passed": passed,
            "value": dup_rate,
            "threshold": "=0",
            "message": f"{columns} 重复率：{dup_rate:.4f}"
        })
        return self

    def check_range(self, column: str, min_val=None, max_val=None):
        """检查数值范围"""
        if min_val is not None:
            below_min = (self.df[column] < min_val).sum()
            passed = below_min == 0
            self.results.append({
                "check": f"range_min_{column}",
                "passed": passed,
                "value": below_min,
                "threshold": f"{column}>={min_val}",
                "message": f"{column} 低于 {min_val} 的记录数：{below_min}"
            })
        if max_val is not None:
            above_max = (self.df[column] > max_val).sum()
            passed = above_max == 0
            self.results.append({
                "check": f"range_max_{column}",
                "passed": passed,
                "value": above_max,
                "threshold": f"{column}<={max_val}",
                "message": f"{column} 高于 {max_val} 的记录数：{above_max}"
            })
        return self

    def get_report(self) -> dict:
        """汇总检查结果"""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])
        failed = total - passed

        return {
            "table": self.table_name,
            "ds": self.ds,
            "total_checks": total,
            "passed": passed,
            "failed": failed,
            "all_passed": failed == 0,
            "details": self.results,
        }

    def assert_all_passed(self):
        """如果有任何检查不通过，抛出异常"""
        report = self.get_report()
        for detail in report["details"]:
            print(f"  {'PASS' if detail['passed'] else 'FAIL'} | {detail['check']}: {detail['message']}")

        if not report["all_passed"]:
            failed_checks = [d for d in report["details"] if not d["passed"]]
            raise AssertionError(
                f"数据质量检查失败！"
                f"{report['failed']}/{report['total_checks']} 项不通过：\n"
                + "\n".join([f"  - {d['check']}: {d['message']}" for d in failed_checks])
            )

# 使用示例
def run_quality_check(**context):
    import pandas as pd
    ds = context["ds"]

    df = pd.read_csv(f"/data/cleaned/orders/dt={ds}/orders.csv")

    checker = DataQualityChecker(df, "ods.orders", ds)
    checker \
        .check_not_empty() \
        .check_null_rate("amount", threshold=0.05) \
        .check_null_rate("user_id", threshold=0.01) \
        .check_uniqueness(["order_id"]) \
        .check_range("amount", min_val=0) \
        .assert_all_passed()

    return checker.get_report()

---

小结

本文演示了一个覆盖主要处理环节的数据管道教学骨架，跨系统同步、部署材料与故障恢复机制仍需补齐，不能直接视为完整的生产级实现：

步骤	关键设计
数据抽取	增量按日期，幂等（覆盖写入）
质量检查	独立 Task，retries=0，失败即告警
数据清洗	标准化、过滤、空值处理
加载到 ODS	先 DROP PARTITION，再 LOAD（重复写入思路，仍需故障恢复保护）
聚合到 DWS	INSERT OVERWRITE（幂等）
同步到 MySQL	先 DELETE 再 INSERT（重复写入思路，仍需事务及恢复验证）
报表邮件	HTML 格式，汇总关键指标

幂等性是最重要的设计原则：每个 Task 重复运行 N 次的结果应该相同，这样才能放心重试和补数。

下一篇：11-Airflow进阶实践 — CeleryExecutor、KubernetesExecutor、自定义 Operator

---

所属系列：00-Airflow学习路线图 | 上一篇：09-监控告警与故障处理 | 下一篇：11-Airflow进阶实践

#Airflow #ETL #数据管道 #增量处理 #数据质量 #MySQL #Hive #实战

---

---

掌握检查

• [ ] 能解释为什么数据质量检查 Task 应该设置 retries=0，而数据抽取 Task 可以设置 retries=2
• [ ] 能说出本案例中三种重复写入处理思路：加载到 ODS 先 DROP PARTITION 再 LOAD DATA、同步到 MySQL 先 DELETE 再 INSERT、Hive 聚合用 INSERT OVERWRITE，并解释前两者为何仍需故障恢复或事务保护
• [ ] 能写出按日期增量抽取 MySQL 数据的 SQL（WHERE 条件使用 {{ ds }}）
• [ ] 理解为什么把数据质量检查放在数据清洗之前（尽早失败，减少无效计算）
• [ ] 知道如何在 DAG 中集成钉钉告警，使任何 Task 失败时自动通知

---

实践练习

基础题：把本文完整的 ecommerce_daily_etl DAG 结构用 mermaid 流程图画出来（不需要代码，只画 Task 节点和依赖箭头），标注每个 Task 的 Operator 类型，以及哪个 Task 是并行、哪个是串行。

进阶题：假设数据质量检查发现某天数据量为 0（不是系统故障，而是业务侧真的没有数据），但你不想让整个 DAG 失败、下游所有 Task 都被跳过。如何在数据质量检查 Task 中处理这种"合理的空数据"情况？给出具体的代码思路。