Airflow数据血缘关系追踪指南：如何追溯上游数据源与DAG依赖

tracing-upstream-lineage by astronomer/agents

469 周安装量

292 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/astronomer/agents --skill tracing-upstream-lineage

数据分析开发运维数据处理

🇨🇳中文介绍

上游血缘关系：数据源

追溯数据来源——回答“这些数据从何而来？”

血缘关系调查

步骤 1：确定目标类型

确定我们要追踪的目标：

表：追踪填充此表的数据来源
列：追踪此特定列的来源
DAG ：追踪此 DAG 读取的数据源

步骤 2：查找生产 DAG

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

按名称搜索 DAG ：使用 af dags list 并查找与表名匹配的 DAG 名称
- load_customers -> customers 表
- etl_daily_orders -> orders 表
探索 DAG 源代码 ：使用读取 DAG 定义

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

步骤 3：追踪数据源

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

SQL 源（查找 FROM 子句）：

# In DAG code:
SELECT * FROM source_schema.source_table  # <- 这是一个上游源

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

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 代码中记录连接信息

当追踪特定列时：

在目标表模式中找到该列
在 DAG 源代码中搜索对该列名的引用
追踪转换过程：
- 直接映射：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

描述数据如何流动和转换：

原始数据通过 Salesforce API 同步进入 raw.orders
DAG transform_orders 清理并去重数据到 stg.orders
DAG build_order_facts 与维度表关联后进入 fct.orders

是否存在单点故障？
是否存在过时的上游源？
是否存在可能中断的复杂转换链？

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

🇺🇸English

Upstream Lineage: Sources

Trace the origins of data - answer "Where does this data come from?"

Lineage Investigation

Step 1: Identify the Target Type

Determine what we're tracing:

Table : Trace what populates this table
Column : Trace where this specific column comes from
DAG : Trace what data sources this DAG reads from

Step 2: Find the Producing DAG

Tables are typically populated by Airflow DAGs. Find the connection:

Search DAGs by name : Use af dags list and look for DAG names matching the table name
- load_customers -> customers table
- etl_daily_orders -> orders table
Explore DAG source code : Use af dags source <dag_id> to read the DAG definition
- Look for INSERT, MERGE, CREATE TABLE statements
- Find the target table in the code
Check DAG tasks : Use af tasks list <dag_id> to see what operations the DAG performs

On Astro

If you're running on Astro, the Lineage tab in the Astro UI provides visual lineage exploration across DAGs and datasets. Use it to quickly trace upstream dependencies without manually searching DAG source code.

On OSS Airflow

Use DAG source code and task logs to trace lineage (no built-in cross-DAG UI).

Step 3: Trace Data Sources

From the DAG code, identify source tables and systems:

SQL Sources (look for FROM clauses):

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

External Sources (look for connection references):

S3Operator -> S3 bucket source
PostgresOperator -> Postgres database source
SalesforceOperator -> Salesforce API source
HttpOperator -> REST API source

File Sources :

CSV/Parquet files in object storage
SFTP drops
Local file paths

Step 4: Build the Lineage Chain

Recursively trace each source:

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)

Step 5: Check Source Health

For each upstream source:

Tables : Check freshness with the checking-freshness skill
DAGs : Check recent run status with af dags stats
External systems : Note connection info from DAG code

Lineage for Columns

When tracing a specific column:

Find the column in the target table schema
Search DAG source code for references to that column name
Trace through transformations:
- Direct mappings: source.col AS target_col
- Transformations: COALESCE(a.col, b.col) AS target_col
- Aggregations: SUM(detail.amount) AS total_amount

Output: Lineage Report

Summary

One-line answer: "This table is populated by DAG X from sources Y and Z"

Lineage Diagram

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

Source Details

Source	Type	Connection	Freshness	Owner
raw.orders	Table	Internal	2h ago	data-team
Salesforce	API	salesforce_conn	Real-time	sales-ops

Transformation Chain

Describe how data flows and transforms:

Raw data lands in raw.orders via Salesforce API sync
DAG transform_orders cleans and dedupes into stg.orders
DAG build_order_facts joins with dimensions into fct.orders

Data Quality Implications

Single points of failure?
Stale upstream sources?
Complex transformation chains that could break?

Related Skills

Check source freshness: checking-freshness skill
Debug source DAG: debugging-dags skill
Trace downstream impacts: tracing-downstream-lineage skill
Add manual lineage annotations: annotating-task-lineage skill
Build custom lineage extractors: creating-openlineage-extractors skill

Weekly Installs

388

Repository

astronomer/agents

GitHub Stars

269

First Seen

Jan 23, 2026

Security Audits

Gen Agent Trust HubPass SocketPass SnykPass

Installed on

opencode284

codex275

cursor271

github-copilot266

gemini-cli252

claude-code249

Azure 升级评估与自动化工具 - 轻松迁移 Functions 计划、托管层级和 SKU

66,100 周安装