tracing-upstream-lineage

上游血缘：来源

追踪数据的来源——回答“这些数据来自哪里？”

血缘调查

步骤 1：确定目标类型

确定要追踪的对象：

• 表：追踪哪些数据会填充此表
• 列：追踪此特定列的来源
• DAG：追踪此 DAG 从哪些数据源读取数据

步骤 2：查找生成数据的 DAG

表通常由 Airflow DAG 填充。查找关联关系：

1. 按名称搜索 DAG：使用 af dags list，查找名称与表名匹配的 DAG

   • load_customers -> customers 表
   • etl_daily_orders -> orders 表

2. 查看 DAG 源代码：使用 af dags source <dag_id> 阅读 DAG 定义

   • 查找 INSERT、MERGE、CREATE TABLE 语句
   • 在代码中找到目标表

3. 检查 DAG 任务：使用 af tasks list <dag_id> 查看 DAG 执行了哪些操作

在 Astro 上

如果你在 Astro 上运行，Astro UI 中的 Lineage 选项卡可直观地探索 DAG 和数据集之间的血缘关系。使用它即可快速追踪上游依赖，无需手动搜索 DAG 源代码。

在 OSS Airflow 上

使用 DAG 源代码和任务日志追踪血缘（没有内置的跨 DAG UI）。

步骤 3：追踪数据来源

从 DAG 代码中识别源表和源系统：

SQL 来源（查找 FROM 子句）：

# In DAG code:
SELECT * FROM source_schema.source_table  # <- This is an upstream source

外部来源（查找连接引用）：

• S3Operator -> S3 存储桶来源
• PostgresOperator -> Postgres 数据库来源
• SalesforceOperator -> Salesforce API 来源
• HttpOperator -> REST API 来源

文件来源：

• 对象存储中的 CSV/Parquet 文件
• SFTP 文件投递
• 本地文件路径

步骤 4：构建血缘链

递归追踪每个来源：

TARGET: analytics.orders_daily
    ^
    +-- DAG: etl_daily_orders
            ^
            +-- SOURCE: raw.orders (table)
            |       ^
            |       +-- DAG: ingest_orders
            |               ^
            |               +-- SOURCE: Salesforce API (external)
            |
            +-- SOURCE: dim.customers (table)
                    ^
                    +-- DAG: load_customers
                            ^
                            +-- SOURCE: PostgreSQL (external DB)

步骤 5：检查来源健康状况

对于每个上游来源：

• 表：使用 checking-freshness 技能检查数据新鲜度
• DAG：使用 af dags stats 检查近期运行状态
• 外部系统：记录 DAG 代码中的连接信息

列级血缘

追踪特定列时：

1. 在目标表架构中找到该列
2. 在 DAG 源代码中搜索对此列名的引用
3. 追踪转换过程：
   • 直接映射：source.col AS target_col
   • 转换：COALESCE(a.col, b.col) AS target_col
   • 聚合：SUM(detail.amount) AS total_amount

输出：血缘报告

摘要

用一句话回答：“此表由 DAG X 使用来源 Y 和 Z 填充”

血缘图

[Salesforce] --> [raw.opportunities] --> [stg.opportunities] --> [fct.sales]
                        |                        |
                   DAG: ingest_sfdc         DAG: transform_sales

来源详情

来源	类型	连接	新鲜度	负责人
raw.orders	表	内部	2 小时前	data-team
Salesforce	API	salesforce_conn	实时	sales-ops

转换链

描述数据的流转和转换过程：

1. 原始数据通过 Salesforce API 同步写入 raw.orders
2. DAG transform_orders 清理并去重数据，写入 stg.orders
3. DAG build_order_facts 将数据与维度表关联，写入 fct.orders

数据质量影响

• 是否存在单点故障？
• 上游来源是否过时？
• 转换链是否复杂到可能发生故障？

相关技能

• 检查来源新鲜度：checking-freshness 技能
• 调试来源 DAG：debugging-dags 技能
• 追踪下游影响：tracing-downstream-lineage 技能
• 添加手动血缘注释：annotating-task-lineage 技能
• 构建自定义血缘提取器：creating-openlineage-extractors 技能