tracing-downstream-lineage

作者: astronomer

追踪下游数据血缘,在修改表或DAG前评估变更影响。通过源代码搜索、视图依赖和BI工具连接识别目标表或DAG的直接消费者,构建完整的依赖树,映射从表到仪表盘再到机器学习模型的所有下游影响。按关键性(关键、高、中、低)对依赖进行分类,以优先安排利益相关者沟通和测试。生成包含风险评估、受影响...的影响报告。

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

Downstream Lineage: Impacts

Answer the critical question: "What breaks if I change this?"

Use this BEFORE making changes to understand the blast radius.

Impact Analysis

Step 1: Identify Direct Consumers

Find everything that reads from this target:

For Tables:

  1. Search DAG source code: Look for DAGs that SELECT from this table

    • Use af dags list to get all DAGs
    • Use af dags source <dag_id> to search for table references
    • Look for: FROM target_table, JOIN target_table
  2. Check for dependent views:

    -- Snowflake
    SELECT * FROM information_schema.view_table_usage
    WHERE table_name = '<target_table>'
    
    -- Or check SHOW VIEWS and search definitions
    
  3. Look for BI tool connections:

    • Dashboards often query tables directly
    • Check for common BI patterns in table naming (rpt_, dashboard_)

On Astro

If you're running on Astro, the Lineage tab in the Astro UI provides visual dependency graphs across DAGs and datasets, making downstream impact analysis faster. It shows which DAGs consume a given dataset and their current status, reducing the need for manual source code searches.

For DAGs:

  1. Check what the DAG produces: Use af dags source <dag_id> to find output tables
  2. Then trace those tables' consumers (recursive)

Step 2: Build Dependency Tree

Map the full downstream impact:

SOURCE: fct.orders
    |
    +-- TABLE: agg.daily_sales --> Dashboard: Executive KPIs
    |       |
    |       +-- TABLE: rpt.monthly_summary --> Email: Monthly Report
    |
    +-- TABLE: ml.order_features --> Model: Demand Forecasting
    |
    +-- DIRECT: Looker Dashboard "Sales Overview"

Step 3: Categorize by Criticality

Critical (breaks production):

  • Production dashboards
  • Customer-facing applications
  • Automated reports to executives
  • ML models in production
  • Regulatory/compliance reports

High (causes significant issues):

  • Internal operational dashboards
  • Analyst workflows
  • Data science experiments
  • Downstream ETL jobs

Medium (inconvenient):

  • Ad-hoc analysis tables
  • Development/staging copies
  • Historical archives

Low (minimal impact):

  • Deprecated tables
  • Unused datasets
  • Test data

Step 4: Assess Change Risk

For the proposed change, evaluate:

Schema Changes (adding/removing/renaming columns):

  • Which downstream queries will break?
  • Are there SELECT * patterns that will pick up new columns?
  • Which transformations reference the changing columns?

Data Changes (values, volumes, timing):

  • Will downstream aggregations still be valid?
  • Are there NULL handling assumptions that will break?
  • Will timing changes affect SLAs?

Deletion/Deprecation:

  • Full dependency tree must be migrated first
  • Communication needed for all stakeholders

Step 5: Find Stakeholders

Identify who owns downstream assets:

  1. DAG owners: Check owners field in DAG definitions
  2. Dashboard owners: Usually in BI tool metadata
  3. Team ownership: Look for team naming patterns or documentation

Output: Impact Report

Summary

"Changing fct.orders will impact X tables, Y DAGs, and Z dashboards"

Impact Diagram

                    +--> [agg.daily_sales] --> [Executive Dashboard]
                    |
[fct.orders] -------+--> [rpt.order_details] --> [Ops Team Email]
                    |
                    +--> [ml.features] --> [Demand Model]

Detailed Impacts

DownstreamTypeCriticalityOwnerNotes
agg.daily_salesTableCriticaldata-engUpdated hourly
Executive DashboardDashboardCriticalanalyticsCEO views daily
ml.order_featuresTableHighml-teamRetraining weekly

Risk Assessment

Change TypeRisk LevelMitigation
Add columnLowNo action needed
Rename columnHighUpdate 3 DAGs, 2 dashboards
Delete columnCriticalFull migration plan required
Change data typeMediumTest downstream aggregations

Recommended Actions

Before making changes:

  1. Notify owners: @data-eng, @analytics, @ml-team
  2. Update downstream DAG: transform_daily_sales
  3. Test dashboard: Executive KPIs
  4. Schedule change during low-impact window

Related Skills

  • Trace where data comes from: tracing-upstream-lineage skill
  • Check downstream freshness: checking-freshness skill
  • Debug any broken DAGs: debugging-dags skill
  • Add manual lineage annotations: annotating-task-lineage skill
  • Build custom lineage extractors: creating-openlineage-extractors skill

来自 astronomer 的更多技能

airflow
astronomer
查询、管理和排查Apache Airflow的DAG、运行记录、任务及系统配置。支持30多种命令,涵盖DAG检查、运行管理、任务日志、配置查询及直接REST API访问。通过持久化配置管理多个Airflow实例;自动发现本地和Astro部署。同步(等待完成)或异步触发DAG运行,诊断故障,清除运行记录以重试,并通过重试/映射索引过滤访问任务日志。输出...
official
airflow-hitl
astronomer
在Airflow DAG中使用可延迟操作符实现人工审批关卡、表单输入和分支。四种操作符类型:用于批准/拒绝决策的ApprovalOperator、带表单的多选项选择HITLOperator、人工驱动的任务路由HITLBranchOperator,以及表单数据收集HITLEntryOperator。所有操作符均为可延迟设计,在通过Airflow UI的"必需操作"标签页或REST API等待人工响应时释放工作槽位。支持包括自定义在内的可选功能...
official
airflow-plugins
astronomer
构建嵌入FastAPI应用、自定义UI页面、React组件、中间件、宏和操作符链接的Airflow 3.1+插件,直接集成到Airflow UI中。使用…
official
analyzing-data
astronomer
查询数据仓库,利用缓存的模式和概念映射来回答业务问题。支持对重复问题类型进行模式查找和缓存,并通过记录结果来改进后续查询。包含概念到表的映射缓存,以及通过INFORMATION_SCHEMA或代码库grep进行表结构发现。提供run_sql()和run_sql_pandas()内核函数,返回Polars或Pandas DataFrame用于分析。提供CLI命令用于管理概念、模式和表缓存,以及...
official
annotating-task-lineage
astronomer
使用入口和出口为Airflow任务标注数据血缘。支持使用OpenLineage Dataset对象、Airflow Assets和Airflow Datasets定义跨数据库、数据仓库及云存储的输入输出。当运算符缺少内置OpenLineage提取器时作为备用方案;遵循四级优先级系统,其中自定义提取器和OpenLineage方法优先。包含针对Snowflake、BigQuery、S3和PostgreSQL的数据集命名辅助工具,以确保一致性...
official
authoring-dags
astronomer
创建Apache Airflow DAG的引导式工作流,集成验证与测试。采用六阶段结构化方法:发现环境与现有模式、规划DAG结构、遵循最佳实践实现、通过af CLI命令验证、经用户同意测试、迭代修复。用于发现(af config connections、af config providers、af dags list)和验证(af dags errors、af dags get、af dags explore)的CLI命令可提供DAG的即时反馈...
official
blueprint
astronomer
使用Pydantic验证定义可复用的Airflow任务组模板,并从YAML组合DAG。适用于创建blueprint模板、从YAML组合DAG等场景。
official
checking-freshness
astronomer
通过检查表时间戳和更新模式与陈旧度标尺对比,验证数据新鲜度。使用常见ETL命名模式(如_loaded_at、_updated_at、created_at等)识别时间戳列,并查询其最大值以确定数据时效。将数据分为四种新鲜度状态:新鲜(<4小时)、陈旧(4–24小时)、非常陈旧(>24小时)或未知(未找到时间戳)。提供SQL模板,用于检查最近几天的最后更新时间及行数变化趋势。
official