Data Lineage: Tracing a Column Back to Its Source

Build a small lineage graph for a four-stage pipeline, then use it to trace one output column all the way back to its source file, and to answer “what breaks if this source changes?” before it actually does.

A tracker: a dependency graph plus a column map

class LineageTracker:
    def __init__(self):
        self.edges = {}  # output_name -> {"op": str, "column_map": {out_col: [(source_table, source_col), ...]}}

    def record(self, output, op, column_map):
        self.edges[output] = {"op": op, "column_map": column_map}

    def trace_column(self, table, column, depth=0):
        pad = "  " * depth
        if table not in self.edges:
            print(f"{pad}{table}.{column}  <- source (no further lineage)")
            return
        edge = self.edges[table]
        sources = edge["column_map"].get(column, [])
        src_desc = ", ".join(f"{t}.{c}" for t, c in sources)
        print(f"{pad}{table}.{column}  <- {edge['op']} <- {src_desc}")
        for src_table, src_col in sources:
            self.trace_column(src_table, src_col, depth + 1)

    def inputs_of(self, output):
        return sorted({t for cols in self.edges[output]["column_map"].values() for t, _ in cols})

Recording lineage for a four-stage pipeline

lineage = LineageTracker()

lineage.record(output="bronze.orders", op="ingest (as-is)", column_map={
    "order_id": [("landing/orders.csv", "order_id")],
    "amount": [("landing/orders.csv", "amount")],
    "customer_id": [("landing/orders.csv", "customer_id")],
})
lineage.record(output="bronze.customers", op="ingest (as-is)", column_map={
    "country_code": [("landing/customers.csv", "country_code")],
})
lineage.record(output="silver.orders", op="filter status != pending", column_map={
    "order_id": [("bronze.orders", "order_id")],
    "amount": [("bronze.orders", "amount")],
    "customer_id": [("bronze.orders", "customer_id")],
})
lineage.record(output="gold.daily_revenue", op="join + group by country, sum(amount)", column_map={
    "revenue": [("silver.orders", "amount")],
    "country_code": [("bronze.customers", "country_code")],
})

Tracing one output column back to its source

print("== tracing gold.daily_revenue.revenue back to its source ==")
lineage.trace_column("gold.daily_revenue", "revenue")

$ python lineage.py
== tracing gold.daily_revenue.revenue back to its source ==
gold.daily_revenue.revenue  <- join + group by country, sum(amount) <- silver.orders.amount
  silver.orders.amount  <- filter status != pending <- bronze.orders.amount
    bronze.orders.amount  <- ingest (as-is) <- landing/orders.csv.amount
      landing/orders.csv.amount  <- source (no further lineage)

Four hops, and the chain is unambiguous: revenue in the gold table came from exactly one column, amount, in exactly one source file. Column-level lineage is what makes that “exactly” possible — table-level lineage would only be able to say gold.daily_revenue depends on silver.orders and bronze.customers, without saying which one actually produced revenue.

print("== tracing gold.daily_revenue.country_code back to its source ==")
lineage.trace_column("gold.daily_revenue", "country_code")

== tracing gold.daily_revenue.country_code back to its source ==
gold.daily_revenue.country_code  <- join + group by country, sum(amount) <- bronze.customers.country_code
  bronze.customers.country_code  <- ingest (as-is) <- landing/customers.csv.country_code
    landing/customers.csv.country_code  <- source (no further lineage)

Same output table, different column, completely different source chain — revenue traces through bronze.orders, country_code traces through bronze.customers entirely. A naive tracker that just recorded “gold.daily_revenue depends on silver.orders and bronze.customers” without per-column detail would get this indistinguishable, and a debugging session chasing a wrong country_code value would have to check both upstream tables instead of just one.

Impact analysis: what breaks before it breaks

print("== impact analysis: what breaks if landing/customers.csv changes shape ==")
affected = [out for out in lineage.edges if "landing/customers.csv" in lineage.inputs_of(out)]
downstream = list(affected)
frontier = affected
while frontier:
    next_frontier = [out for out in lineage.edges
                      if any(f in lineage.inputs_of(out) for f in frontier) and out not in downstream]
    downstream.extend(next_frontier)
    frontier = next_frontier
print(f"landing/customers.csv changing affects: {downstream}")

== impact analysis: what breaks if landing/customers.csv changes shape ==
landing/customers.csv changing affects: ['bronze.customers', 'gold.daily_revenue']

Walking the graph forward from a source instead of backward from an output answers the other half of the lineage question: before renaming a column in landing/customers.csv, this says exactly which two downstream tables need updating, and — because it’s column-level — that bronze.orders and silver.orders are entirely unaffected, with no need to check them by hand.

Practice

gold.order_count.count  <- count(*) <- silver.orders.order_id
  silver.orders.order_id  <- filter status != pending <- bronze.orders.order_id
    bronze.orders.order_id  <- ingest (as-is) <- landing/orders.csv.order_id
      landing/orders.csv.order_id  <- source (no further lineage)

landing/orders.csv changing affects: ['bronze.orders', 'silver.orders', 'gold.daily_revenue']

gold.daily_revenue.margin  <- join + group by country, sum(amount) <- 

Next: data governance — access control, cataloging, and policy enforcement on top of the tables this lineage graph describes.