armor-lineage

Data Lineage

Explore upstream dependencies and downstream consumers of your data assets.

Prerequisites

• AnomalyArmor API key configured (~/.armor/config.yaml or ARMOR_API_KEY env var)
• Python SDK installed (pip install anomalyarmor)

When to Use

• "What depends on this table?"
• "Where does this data come from?"
• "Impact analysis for changes"
• "Show upstream dependencies"
• "List downstream consumers"
• "Trace data flow"

Concepts

Directions

• upstream: Tables that this table depends on (data sources)
• downstream: Tables that depend on this table (data consumers)
• both: Full lineage graph in both directions

Depth

• depth=1: Direct dependencies only
• depth=2: Two levels of dependencies
• depth=3+: Extended dependency chain (max 5)

Steps

1. Identify the asset to analyze
2. Determine direction (upstream, downstream, or both)
3. Choose appropriate depth (start with 1-2)
4. Call client.lineage.get() to fetch lineage graph
5. Analyze the graph for impact or root cause

Example Usage

Find Upstream Dependencies

from anomalyarmor import Client

client = Client()

# Get upstream lineage (where data comes from)
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="upstream",
    depth=2
)

print(f"Table: {lineage.root.qualified_name}")
print(f"\nUpstream Dependencies ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
    print(f"  {node.qualified_name}")
    if node.asset_type:
        print(f"    Type: {node.asset_type}")

Find Downstream Consumers

# Get downstream lineage (what depends on this)
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="downstream",
    depth=2
)

print(f"Table: {lineage.root.qualified_name}")
print(f"\nDownstream Consumers ({len(lineage.downstream)} tables):")
for node in lineage.downstream:
    print(f"  {node.qualified_name}")

Impact Analysis

# Full impact analysis before making changes
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="both",
    depth=3
)

print("=== IMPACT ANALYSIS ===")
print(f"\nTable: {lineage.root.qualified_name}")

# Upstream (what feeds this table)
print(f"\nData Sources ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
    print(f"  <- {node.qualified_name}")

# Downstream (what will be affected by changes)
print(f"\nWill Impact ({len(lineage.downstream)} tables):")
for node in lineage.downstream:
    print(f"  -> {node.qualified_name}")

# Edges show the relationships
print(f"\nRelationships ({len(lineage.edges)} edges):")
for edge in lineage.edges:
    print(f"  {edge.source} -> {edge.target}")

List All Lineage

# List lineage summaries for multiple assets
lineage_list = client.lineage.list(limit=50)
for item in lineage_list:
    print(f"{item.qualified_name}:")
    print(f"  Upstream: {item.upstream_count}")
    print(f"  Downstream: {item.downstream_count}")

Expected Output

=== IMPACT ANALYSIS ===

Table: warehouse.gold.fact_orders

Data Sources (4 tables):
  <- warehouse.staging.orders_raw
  <- warehouse.staging.customers
  <- warehouse.staging.products
  <- warehouse.raw.events

Will Impact (7 tables):
  -> warehouse.gold.daily_revenue
  -> warehouse.gold.customer_metrics
  -> warehouse.reporting.exec_dashboard
  -> warehouse.reporting.sales_report
  -> warehouse.ml.churn_features
  -> warehouse.ml.ltv_predictions
  -> external.bi_tool.orders_view

Relationships (11 edges):
  staging.orders_raw -> gold.fact_orders
  staging.customers -> gold.fact_orders
  ...

Follow-up Actions

• For upstream issues: Investigate source tables with /armor:investigate
• For impact analysis: Notify downstream consumers of planned changes
• For data quality: Set up monitoring with /armor:quality on critical paths
• For schema changes: Review lineage before applying changes
• To understand data flow: Export lineage to documentation