Data Lineage and Risk Scoring Engine

Build a Python CLI that traces column-level data lineage and scores PII risk.

Introduction

30 Second Summary

When a data breach hits, the first question regulators ask is "which systems did that data touch?" Most organizations struggle to answer because they have no map of how sensitive data flows through their pipelines.

In this project, you will build a Python CLI tool that traces every column through a simulated data pipeline. It detects sensitive data, scores each column for risk, and generates governance-ready reports with quantified metrics.

What You'll Build

A column-level data lineage and risk scoring engine that models a 6-stage data pipeline as a directed acyclic graph, detects PII across 7 categories, and outputs quantified risk reports in Markdown and JSON.

By the end of this project, you'll have:

  • A working column-level lineage graph with 38 nodes and 26 edges, tracking data flow across 6 pipeline stages using NetworkX.
  • A multi-dimensional risk scoring engine that detects PII columns, computes sensitivity, propagation, and exposure scores, and outputs a composite risk score (0.0-1.0) for every column.
  • Auto-generated governance deliverables including a Markdown lineage report with 10+ numerical metrics and a JSON risk summary, ready for stakeholder review.
  • Secret Mission: add interactive impact queries and a DOT graph export with risk-tier color coding for visual lineage documentation.

Are there any prerequisites?

This project assumes you are comfortable with advanced Python concepts like classes, dictionaries, list comprehensions, and regex. A basic understanding of how data moves through pipeline stages is helpful but not required.

Before We Start

When a data breach hits, the governance team needs to know exactly which downstream reports and dashboards were affected.

Answering that question requires column-level data lineage, a map that traces every field from its source through every transformation to its final destination. In this step, you'll create the project folder and install the two Python packages that power the lineage engine.

In this step, get ready to:
  • Create the lineage-engine project folder with data and output subdirectories.
  • Install networkx and pandas.
  • Verify both packages are working.
Set up the project folder
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and its subdirectories by running these commands:
mkdir lineage-engine
cd lineage-engine
mkdir data output

What do these folders do?

The data folder will hold the sample CSV files the pipeline simulator generates.

The output folder will hold the governance reports the engine produces, including a Markdown lineage report and a JSON risk summary.

  • Confirm the folders exist by running:
ls

You should see data and output listed in the output.

Seeing a "file exists" error?

If you already have a folder called lineage-engine on your Desktop from a previous attempt, delete it first and re-run the commands above.

help me fix a folder creation error in my terminal

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the lineage-engine folder on your Desktop, then click Open.

Using a different code editor?

Any code editor works for this project. Open your preferred editor and navigate to the lineage-engine folder on your Desktop.

Install and verify dependencies

The lineage engine relies on two Python packages. Both require Python 3.11 or higher.

  • Install both packages by running this command in your terminal:
pip install networkx==3.6.1 pandas==3.0.3

What are these packages?

NetworkX is a Python library for creating and analyzing graphs (networks of nodes and edges). You'll use it to build the directed graph that models column-level data lineage.

pandas is a data analysis library. You'll use it to generate realistic sample CSV data for the pipeline simulator.

  • Verify both packages installed correctly by running:
python -c "import networkx; import pandas; print(networkx.__version__, pandas.__version__)"

✔️ I see 3.6.1 3.0.3

Both packages are installed and ready to go.

ⓧ I see a "requires a different Python" error

This means your default python or pip is pointing to a Python version below 3.11. Both networkx 3.6.1 and pandas 3.0.3 require Python 3.11 or higher.

  • Check which Python versions you have installed by running:
python3 --version

If this shows 3.11 or higher, use python3 and pip3 for all commands throughout this project instead of python and pip.

  • Re-run the install with:
pip3 install networkx==3.6.1 pandas==3.0.3

On macOS, the system python3 may point to an older version. If python3 --version shows below 3.11, install Python 3.11 or higher from python.org and use the full path (e.g. python3.11) or a version manager like pyenv.

ⓧ I see a different error

A few things to check:

  • Make sure you ran the pip install command first. The verify command only works after both packages are installed.
  • Check for typos in the verify command. The version attributes use double underscores on both sides: __version__.
  • If you see a "permission denied" error, try adding --user to the install command.

help me troubleshoot my pip install error

Your environment is ready. Next up, you'll define the six pipeline stages and build the column-level lineage graph that maps every data flow.

Build the Pipeline Simulator

Your project environment is set up. Now you need something to trace lineage through.

In this step you will build a pipeline simulator that models a realistic 6-stage data pipeline. The simulator defines every column at every stage, maps exactly which source columns feed into each destination column, and generates sample CSV data with PII-like fields such as SSNs, emails, and card numbers.

In this step, get ready to:
  • Define 6 pipeline stages with explicit column-to-column lineage mappings.
  • Build a data generator that produces 50 customer records and 200 transaction records as CSVs.
  • Verify the simulator outputs the expected data files.
Define the pipeline stages

A data pipeline moves information through a series of stages. Each stage takes columns in, transforms or combines them, and passes columns out to the next stage.

To track lineage, you need metadata that records which column at which stage came from which column at a prior stage. You will store this metadata in a Python dictionary.

  • In your code editor, create a new file called pipeline.py in the lineage-engine folder.
  • Add the module docstring and imports by pasting this code at the top of the file:
"""Data pipeline simulator - generates sample data and defines transformation metadata."""

import os
import random
import string
from datetime import datetime, timedelta

import pandas as pd

What do these imports do?

  • os handles file paths and directory creation.
  • random and string generate fake PII data like SSNs and card numbers.
  • datetime and timedelta create realistic dates of birth and transaction dates.
  • pandas converts the generated data into DataFrames and writes them to CSV files.
  • Add the pipeline metadata dictionary below the imports. This is the core data structure that maps every column through the pipeline:


PIPELINE_STAGES = {
    "raw_customers": {
        "type": "source",
        "columns": [
            "customer_id", "first_name", "last_name", "email",
            "ssn", "phone", "date_of_birth", "address",
        ],
    },
    "raw_transactions": {
        "type": "source",
        "columns": [
            "transaction_id", "customer_id", "amount",
            "merchant", "transaction_date", "card_number",
        ],
    },
    "cleaned_customers": {
        "type": "transform",
        "columns": [
            "customer_id", "full_name", "email",
            "ssn_masked", "phone", "age",
        ],
        "lineage": {
            "customer_id": [("raw_customers", "customer_id")],
            "full_name": [("raw_customers", "first_name"), ("raw_customers", "last_name")],
            "email": [("raw_customers", "email")],
            "ssn_masked": [("raw_customers", "ssn")],
            "phone": [("raw_customers", "phone")],
            "age": [("raw_customers", "date_of_birth")],
        },
    },
    "enriched_transactions": {
        "type": "transform",
        "columns": [
            "transaction_id", "customer_id", "full_name",
            "email", "amount", "merchant", "transaction_date",
        ],
        "lineage": {
            "transaction_id": [("raw_transactions", "transaction_id")],
            "customer_id": [
                ("raw_transactions", "customer_id"),
                ("cleaned_customers", "customer_id"),
            ],
            "full_name": [("cleaned_customers", "full_name")],
            "email": [("cleaned_customers", "email")],
            "amount": [("raw_transactions", "amount")],
            "merchant": [("raw_transactions", "merchant")],
            "transaction_date": [("raw_transactions", "transaction_date")],
        },
    },
    "customer_risk_profile": {
        "type": "aggregate",
        "columns": [
            "customer_id", "full_name", "total_spent",
            "avg_transaction", "transaction_count", "risk_flag",
        ],
        "lineage": {
            "customer_id": [("enriched_transactions", "customer_id")],
            "full_name": [("enriched_transactions", "full_name")],
            "total_spent": [("enriched_transactions", "amount")],
            "avg_transaction": [("enriched_transactions", "amount")],
            "transaction_count": [("enriched_transactions", "transaction_id")],
            "risk_flag": [("enriched_transactions", "amount")],
        },
    },
    "monthly_report": {
        "type": "report",
        "columns": [
            "customer_id", "full_name", "total_spent",
            "risk_flag", "last_transaction_date",
        ],
        "lineage": {
            "customer_id": [("customer_risk_profile", "customer_id")],
            "full_name": [("customer_risk_profile", "full_name")],
            "total_spent": [("customer_risk_profile", "total_spent")],
            "risk_flag": [("customer_risk_profile", "risk_flag")],
            "last_transaction_date": [("enriched_transactions", "transaction_date")],
        },
    },
}

What does this dictionary define?

PIPELINE_STAGES is a dictionary where each key is a pipeline stage name and each value describes that stage's structure. There are four stage types that model how data flows in a real pipeline:

  • source stages (raw_customers, raw_transactions) are where data enters the pipeline. They list columns but have no lineage key because nothing feeds into them.
  • transform stages (cleaned_customers, enriched_transactions) reshape data. Their lineage dict maps each destination column to a list of (source_stage, source_column) tuples.
  • aggregate (customer_risk_profile) rolls up transaction data into per-customer summaries.
  • report (monthly_report) is the final output that pulls from both aggregated and enriched data.

Notice that full_name in cleaned_customers traces back to both first_name and last_name in raw_customers. This is how column-level lineage captures merges.

  • Save pipeline.py.

You won't see a change yet. The next chunk adds the data generator that brings these stage definitions to life.

Seeing a syntax error?

  • Check that every opening parenthesis, bracket, and brace has a matching close. Nested tuples inside lists inside dicts are easy to mismatch.
  • Make sure every string uses matching quote marks and every list item ends with a comma.

Ask for help: Why am I getting a SyntaxError in my PIPELINE_STAGES dictionary?

Add the sample data generator

The pipeline metadata describes the structure, but you also need realistic data flowing through it. The generator creates sample customer and transaction CSVs populated with PII-like fields.

  • Add the generate_sample_data function below the PIPELINE_STAGES dictionary:


def generate_sample_data(output_dir="data"):
    """Generate sample CSV files for the pipeline source stages."""
    os.makedirs(output_dir, exist_ok=True)
    random.seed(42)

    customers = []
    for i in range(1, 51):
        customers.append({
            "customer_id": f"CUST-{i:04d}",
            "first_name": f"First{i}",
            "last_name": f"Last{i}",
            "email": f"user{i}@example.com",
            "ssn": f"{random.randint(100, 999)}-{random.randint(10, 99)}-{random.randint(1000, 9999)}",
            "phone": f"+1-{random.randint(200, 999)}-{random.randint(100, 999)}-{random.randint(1000, 9999)}",
            "date_of_birth": (
                datetime(1960, 1, 1) + timedelta(days=random.randint(0, 15000))
            ).strftime("%Y-%m-%d"),
            "address": f"{random.randint(1, 9999)} Main St, City{i}, ST {random.randint(10000, 99999)}",
        })

    df_customers = pd.DataFrame(customers)
    df_customers.to_csv(os.path.join(output_dir, "raw_customers.csv"), index=False)

    transactions = []
    for i in range(1, 201):
        cust_id = random.randint(1, 50)
        transactions.append({
            "transaction_id": f"TXN-{i:06d}",
            "customer_id": f"CUST-{cust_id:04d}",
            "amount": round(random.uniform(5.0, 5000.0), 2),
            "merchant": random.choice([
                "Amazon", "Walmart", "Target", "BestBuy", "Costco", "HomeDepot",
            ]),
            "transaction_date": (
                datetime(2024, 1, 1) + timedelta(days=random.randint(0, 365))
            ).strftime("%Y-%m-%d"),
            "card_number": "".join(random.choices(string.digits, k=16)),
        })

    df_transactions = pd.DataFrame(transactions)
    df_transactions.to_csv(os.path.join(output_dir, "raw_transactions.csv"), index=False)

    return {
        "raw_customers": df_customers,
        "raw_transactions": df_transactions,
    }

What does this function do?

generate_sample_data creates two CSV files that simulate the raw data entering the pipeline:

  • random.seed(42) locks the random number generator to a fixed starting point. This makes the generated data identical every time you run the function.
  • The customer loop creates 50 records, each with PII fields like SSN, phone, date of birth, and address.
  • The transaction loop creates 200 records, each linked to a random customer and populated with amounts, merchants, dates, and a 16-digit card number.
  • to_csv(..., index=False) writes each DataFrame to a CSV without adding a row index column.

The function returns a dictionary of both DataFrames so you can inspect them programmatically later.

  • Save pipeline.py.

Seeing an IndentationError?

  • Make sure the def generate_sample_data line starts at the leftmost indentation level (no leading spaces). It should sit at the same level as PIPELINE_STAGES.
  • Confirm your editor uses spaces, not tabs. Mixing tabs and spaces causes indentation errors in Python.

Ask for help: Why am I getting an IndentationError in pipeline.py?

✔️ Awesome, I've got everything!

Great. Double check you have saved pipeline.py before moving on.

ⓧ I'd like to double check the full code

"""Data pipeline simulator - generates sample data and defines transformation metadata."""

import os
import random
import string
from datetime import datetime, timedelta

import pandas as pd


PIPELINE_STAGES = {
    "raw_customers": {
        "type": "source",
        "columns": [
            "customer_id", "first_name", "last_name", "email",
            "ssn", "phone", "date_of_birth", "address",
        ],
    },
    "raw_transactions": {
        "type": "source",
        "columns": [
            "transaction_id", "customer_id", "amount",
            "merchant", "transaction_date", "card_number",
        ],
    },
    "cleaned_customers": {
        "type": "transform",
        "columns": [
            "customer_id", "full_name", "email",
            "ssn_masked", "phone", "age",
        ],
        "lineage": {
            "customer_id": [("raw_customers", "customer_id")],
            "full_name": [("raw_customers", "first_name"), ("raw_customers", "last_name")],
            "email": [("raw_customers", "email")],
            "ssn_masked": [("raw_customers", "ssn")],
            "phone": [("raw_customers", "phone")],
            "age": [("raw_customers", "date_of_birth")],
        },
    },
    "enriched_transactions": {
        "type": "transform",
        "columns": [
            "transaction_id", "customer_id", "full_name",
            "email", "amount", "merchant", "transaction_date",
        ],
        "lineage": {
            "transaction_id": [("raw_transactions", "transaction_id")],
            "customer_id": [
                ("raw_transactions", "customer_id"),
                ("cleaned_customers", "customer_id"),
            ],
            "full_name": [("cleaned_customers", "full_name")],
            "email": [("cleaned_customers", "email")],
            "amount": [("raw_transactions", "amount")],
            "merchant": [("raw_transactions", "merchant")],
            "transaction_date": [("raw_transactions", "transaction_date")],
        },
    },
    "customer_risk_profile": {
        "type": "aggregate",
        "columns": [
            "customer_id", "full_name", "total_spent",
            "avg_transaction", "transaction_count", "risk_flag",
        ],
        "lineage": {
            "customer_id": [("enriched_transactions", "customer_id")],
            "full_name": [("enriched_transactions", "full_name")],
            "total_spent": [("enriched_transactions", "amount")],
            "avg_transaction": [("enriched_transactions", "amount")],
            "transaction_count": [("enriched_transactions", "transaction_id")],
            "risk_flag": [("enriched_transactions", "amount")],
        },
    },
    "monthly_report": {
        "type": "report",
        "columns": [
            "customer_id", "full_name", "total_spent",
            "risk_flag", "last_transaction_date",
        ],
        "lineage": {
            "customer_id": [("customer_risk_profile", "customer_id")],
            "full_name": [("customer_risk_profile", "full_name")],
            "total_spent": [("customer_risk_profile", "total_spent")],
            "risk_flag": [("customer_risk_profile", "risk_flag")],
            "last_transaction_date": [("enriched_transactions", "transaction_date")],
        },
    },
}


def generate_sample_data(output_dir="data"):
    """Generate sample CSV files for the pipeline source stages."""
    os.makedirs(output_dir, exist_ok=True)
    random.seed(42)

    customers = []
    for i in range(1, 51):
        customers.append({
            "customer_id": f"CUST-{i:04d}",
            "first_name": f"First{i}",
            "last_name": f"Last{i}",
            "email": f"user{i}@example.com",
            "ssn": f"{random.randint(100, 999)}-{random.randint(10, 99)}-{random.randint(1000, 9999)}",
            "phone": f"+1-{random.randint(200, 999)}-{random.randint(100, 999)}-{random.randint(1000, 9999)}",
            "date_of_birth": (
                datetime(1960, 1, 1) + timedelta(days=random.randint(0, 15000))
            ).strftime("%Y-%m-%d"),
            "address": f"{random.randint(1, 9999)} Main St, City{i}, ST {random.randint(10000, 99999)}",
        })

    df_customers = pd.DataFrame(customers)
    df_customers.to_csv(os.path.join(output_dir, "raw_customers.csv"), index=False)

    transactions = []
    for i in range(1, 201):
        cust_id = random.randint(1, 50)
        transactions.append({
            "transaction_id": f"TXN-{i:06d}",
            "customer_id": f"CUST-{cust_id:04d}",
            "amount": round(random.uniform(5.0, 5000.0), 2),
            "merchant": random.choice([
                "Amazon", "Walmart", "Target", "BestBuy", "Costco", "HomeDepot",
            ]),
            "transaction_date": (
                datetime(2024, 1, 1) + timedelta(days=random.randint(0, 365))
            ).strftime("%Y-%m-%d"),
            "card_number": "".join(random.choices(string.digits, k=16)),
        })

    df_transactions = pd.DataFrame(transactions)
    df_transactions.to_csv(os.path.join(output_dir, "raw_transactions.csv"), index=False)

    return {
        "raw_customers": df_customers,
        "raw_transactions": df_transactions,
    }
Verify the pipeline simulator

Time to confirm the simulator generates the expected CSV files with the right number of rows.

  • Run the pipeline simulator by executing this command in your terminal:
python -c "from pipeline import generate_sample_data; data = generate_sample_data(); print('Customers:', len(data['raw_customers']), 'rows'); print('Transactions:', len(data['raw_transactions']), 'rows')"
  • You should see this output:
Customers: 50 rows
Transactions: 200 rows
  • Confirm the CSV files exist by running:
ls data/
  • You should see raw_customers.csv and raw_transactions.csv in the output.

Seeing ModuleNotFoundError?

  • Make sure your terminal is inside the lineage-engine folder. Python needs to find pipeline.py in the current directory.
  • If you see a pandas import error, confirm you installed the dependencies in the same Python environment by running pip install networkx==3.6.1 pandas==3.0.3 again.

Ask for help: Why is Python unable to find my pipeline module or pandas package?

What is PII and why does this data contain it?

PII stands for Personally Identifiable Information. It includes any data that could identify a specific person, such as Social Security numbers, email addresses, phone numbers, and credit card numbers.

The sample data intentionally includes PII-like fields so the risk scoring engine you build next can detect them, classify their sensitivity, and compute how far that sensitive data propagates through the pipeline.

Your pipeline simulator is ready. Next up, you will build a lineage graph engine that turns these stage definitions into a traversable directed acyclic graph using NetworkX.

Build the Lineage Graph Engine

Your pipeline simulator is ready, with 50 customer records and 200 transactions sitting in CSV files. Now you need a way to trace how each column flows through all 6 stages of that pipeline.

In this step, you'll build a graph engine that reads the lineage metadata from PIPELINE_STAGES and constructs a directed acyclic graph where every node is a column and every edge is a data flow. You'll also add functions to compute depth metrics and summary statistics for the graph.

In this step, get ready to:
  • Create the lineage graph constructor that turns pipeline metadata into a directed graph.
  • Add depth computation and summary metric functions.
  • Verify the graph has the expected node count, edge count, and longest path length.
Create the lineage graph engine

The lineage graph represents each column in the pipeline as a node (like raw_customers.ssn) and each data flow as a directed edge. NetworkX's DiGraph class is built for exactly this kind of structure.

  • Create a new file called lineage.py inside your lineage-engine folder (the same directory where pipeline.py lives).
  • Add the imports and the graph construction function by pasting this code:
"""Lineage graph engine - builds a NetworkX DiGraph for column-level data lineage."""

import networkx as nx

from pipeline import PIPELINE_STAGES


def build_lineage_graph():
    """Build a directed acyclic graph representing column-level data lineage."""
    G = nx.DiGraph()

    for stage_name, stage_info in PIPELINE_STAGES.items():
        stage_type = stage_info["type"]

        for column in stage_info["columns"]:
            node_id = f"{stage_name}.{column}"
            G.add_node(node_id, stage=stage_name, column=column, stage_type=stage_type)

        if "lineage" in stage_info:
            for dest_column, sources in stage_info["lineage"].items():
                dest_node = f"{stage_name}.{dest_column}"
                for source_stage, source_column in sources:
                    source_node = f"{source_stage}.{source_column}"
                    G.add_edge(source_node, dest_node)

    return G

What does this code do?

The function loops through every stage in PIPELINE_STAGES and creates a node for each column. Each node ID follows the format stage.column (e.g. raw_customers.ssn), and stores the stage name, column name, and stage type as node attributes.

For stages that have a lineage dict, the function reads each destination column's source mappings and creates a directed edge from the source node to the destination node. This is what connects raw_customers.ssn to cleaned_customers.ssn_masked, for example.

You won't see a change yet. The next function adds depth computation to the graph.

  • Add the depth computation function below build_lineage_graph by pasting this code at the bottom of lineage.py:
def compute_all_depths(G):
    """Compute depth (longest path from any source) for every node using topological sort."""
    depths = {}
    for node in nx.topological_sort(G):
        preds = list(G.predecessors(node))
        if not preds:
            depths[node] = 0
        else:
            depths[node] = 1 + max(depths[p] for p in preds)
    return depths

What does this function do?

This function computes how "deep" each column is in the pipeline by finding the longest path from any source node to that column. It uses topological sort to process nodes in dependency order, so every predecessor's depth is already known before the current node is computed.

Source columns (like raw_customers.email) get depth 0. A column four stages downstream would get depth 4. This runs in O(V+E) time, making it efficient even for large graphs.

You won't see a change yet. The next function adds summary statistics to the graph.

  • Add the graph metrics function below compute_all_depths by pasting this code at the bottom of lineage.py:
def get_graph_metrics(G):
    """Compute summary metrics for the lineage graph."""
    nodes = G.number_of_nodes()
    edges = G.number_of_edges()

    source_nodes = [n for n in G.nodes() if G.in_degree(n) == 0]
    terminal_nodes = [n for n in G.nodes() if G.out_degree(n) == 0]

    avg_in_degree = round(sum(d for _, d in G.in_degree()) / nodes, 2) if nodes > 0 else 0
    avg_out_degree = round(sum(d for _, d in G.out_degree()) / nodes, 2) if nodes > 0 else 0

    try:
        longest_path = nx.dag_longest_path(G)
        longest_path_length = len(longest_path) - 1 if len(longest_path) > 0 else 0
    except nx.NetworkXUnfeasible:
        longest_path = []
        longest_path_length = 0

    return {
        "total_nodes": nodes,
        "total_edges": edges,
        "source_columns": len(source_nodes),
        "terminal_columns": len(terminal_nodes),
        "avg_in_degree": avg_in_degree,
        "avg_out_degree": avg_out_degree,
        "longest_path_length": longest_path_length,
        "longest_path": [str(n) for n in longest_path],
    }

What does this function do?

This function computes summary statistics about the lineage graph. Source columns are nodes with no incoming edges (the starting points of data flow). Terminal columns are nodes with no outgoing edges (the final destinations).

The nx.dag_longest_path call finds the longest chain of column-to-column flows in the graph. The try/except block catches the edge case where the graph has a cycle (which would mean it's not a valid DAG).

  • Save lineage.py.

Seeing a red underline on the import?

If your editor shows an error on from pipeline import PIPELINE_STAGES, make sure lineage.py is in the same directory as pipeline.py. Both files must be inside the lineage-engine folder.

Help me fix this import error in lineage.py

✔️ Awesome, I've got everything!

Great. Double check you've saved lineage.py before moving on to the verification step.

ⓧ I'd like to double check the full code

"""Lineage graph engine - builds a NetworkX DiGraph for column-level data lineage."""

import networkx as nx

from pipeline import PIPELINE_STAGES


def build_lineage_graph():
    """Build a directed acyclic graph representing column-level data lineage."""
    G = nx.DiGraph()

    for stage_name, stage_info in PIPELINE_STAGES.items():
        stage_type = stage_info["type"]

        for column in stage_info["columns"]:
            node_id = f"{stage_name}.{column}"
            G.add_node(node_id, stage=stage_name, column=column, stage_type=stage_type)

        if "lineage" in stage_info:
            for dest_column, sources in stage_info["lineage"].items():
                dest_node = f"{stage_name}.{dest_column}"
                for source_stage, source_column in sources:
                    source_node = f"{source_stage}.{source_column}"
                    G.add_edge(source_node, dest_node)

    return G


def compute_all_depths(G):
    """Compute depth (longest path from any source) for every node using topological sort."""
    depths = {}
    for node in nx.topological_sort(G):
        preds = list(G.predecessors(node))
        if not preds:
            depths[node] = 0
        else:
            depths[node] = 1 + max(depths[p] for p in preds)
    return depths


def get_graph_metrics(G):
    """Compute summary metrics for the lineage graph."""
    nodes = G.number_of_nodes()
    edges = G.number_of_edges()

    source_nodes = [n for n in G.nodes() if G.in_degree(n) == 0]
    terminal_nodes = [n for n in G.nodes() if G.out_degree(n) == 0]

    avg_in_degree = round(sum(d for _, d in G.in_degree()) / nodes, 2) if nodes > 0 else 0
    avg_out_degree = round(sum(d for _, d in G.out_degree()) / nodes, 2) if nodes > 0 else 0

    try:
        longest_path = nx.dag_longest_path(G)
        longest_path_length = len(longest_path) - 1 if len(longest_path) > 0 else 0
    except nx.NetworkXUnfeasible:
        longest_path = []
        longest_path_length = 0

    return {
        "total_nodes": nodes,
        "total_edges": edges,
        "source_columns": len(source_nodes),
        "terminal_columns": len(terminal_nodes),
        "avg_in_degree": avg_in_degree,
        "avg_out_degree": avg_out_degree,
        "longest_path_length": longest_path_length,
        "longest_path": [str(n) for n in longest_path],
    }
Verify the lineage graph

Your graph engine is built, but you need to confirm that it produces the correct structure. The pipeline metadata should yield exactly 38 nodes (one per column across all 6 stages) and 26 edges (one per lineage mapping).

  • Make sure your terminal is in the lineage-engine directory.
  • Test the graph engine by running this command:
python -c "from lineage import build_lineage_graph, get_graph_metrics; G = build_lineage_graph(); m = get_graph_metrics(G); print('Nodes:', m['total_nodes']); print('Edges:', m['total_edges']); print('Longest path:', m['longest_path_length'])"

You should see this output:

Nodes: 38
Edges: 26
Longest path: 4

What do these numbers mean?

38 nodes means the graph has one node for every column across all 6 pipeline stages (8 + 6 + 6 + 7 + 6 + 5 = 38). 26 edges represents every column-to-column data flow defined in the lineage mappings.

A longest path of 4 means data travels through at most 4 edges (5 nodes) from a raw source column to a final report column. This is the deepest chain of transformations in your pipeline.

Seeing a ModuleNotFoundError?

If the error mentions pipeline, check that your terminal's working directory is the lineage-engine folder where both pipeline.py and lineage.py live. If the error mentions networkx, make sure you're using the same Python environment where you installed the packages in Step 1.

Help me fix a ModuleNotFoundError when running my lineage graph test

Numbers don't match?

If you see fewer nodes or edges, double check that your pipeline.py has all 6 stages with the correct column lists. Each stage's columns list and lineage dict must match the original code exactly.

My lineage graph node or edge count is wrong. Help me debug.

Your lineage graph is capturing every column's journey through the pipeline. Next up, you'll build the risk scoring engine that classifies each column for PII and computes a composite risk score.

Implement PII Detection and Risk Scoring

Your lineage graph is built and verified with 38 nodes and 26 edges. Now you have the structural foundation to ask the critical governance question: which of those columns carry sensitive data, and how far does that sensitivity spread?

In this step, you will build a PII detection engine that classifies columns by name and a multi-dimensional risk scorer. The scorer quantifies sensitivity, propagation, and exposure into a single composite risk score between 0.0 and 1.0.

In this step, get ready to:
  • Classify pipeline columns against 7 PII types using regex pattern matching.
  • Compute propagation risk, exposure score, and composite risk for every column.
  • Score all 38 columns and verify PII detection results.
Create the PII classification patterns and classifier

Before you can score risk, you need to know which columns contain personally identifiable information. The approach here is regex-based name matching: the column name itself tells you whether it holds an SSN, an email, a phone number, and so on.

Why regex name matching instead of content scanning?

Production PII scanners like Google Cloud's Sensitive Data Protection inspect actual cell values. That works for unstructured data, but column-level lineage metadata already carries strong naming conventions.

Matching against column names is fast, deterministic, and works even when you don't have access to the underlying data. This is the same pattern governance teams use to build automated classification rules at scale.

  • Create a new file called risk_scorer.py inside your lineage-engine folder and add the imports and PII patterns configuration:
"""PII detection and multi-dimensional risk scoring engine."""

import re

import networkx as nx

from lineage import compute_all_depths


# Each PII type has regex patterns to match column names,
# a sensitivity score (0.0-1.0), and a risk tier
PII_PATTERNS = {
    "SSN": {
        "column_patterns": [r"ssn", r"social_security"],
        "sensitivity": 1.0,
        "tier": "HIGH",
    },
    "EMAIL": {
        "column_patterns": [r"email", r"e_mail"],
        "sensitivity": 0.7,
        "tier": "HIGH",
    },
    "PHONE": {
        "column_patterns": [r"phone", r"mobile", r"telephone"],
        "sensitivity": 0.6,
        "tier": "MEDIUM",
    },
    "NAME": {
        "column_patterns": [r"first_name", r"last_name", r"full_name"],
        "sensitivity": 0.5,
        "tier": "MEDIUM",
    },
    "DOB": {
        "column_patterns": [r"date_of_birth", r"dob", r"birth_date"],
        "sensitivity": 0.7,
        "tier": "HIGH",
    },
    "ADDRESS": {
        "column_patterns": [r"address", r"street", r"zip_code"],
        "sensitivity": 0.6,
        "tier": "MEDIUM",
    },
    "CARD_NUMBER": {
        "column_patterns": [r"card_number", r"card_num", r"credit_card"],
        "sensitivity": 1.0,
        "tier": "HIGH",
    },
}

What does this code do?

The PII_PATTERNS dictionary defines 7 PII types. Each entry contains regex patterns that match column names, a sensitivity score from 0.0 to 1.0, and a tier label (HIGH, MEDIUM, or NONE).

SSN and CARD_NUMBER carry the highest sensitivity (1.0) because they directly enable identity theft. EMAIL and DOB sit at 0.7 because they are uniquely identifying but less immediately exploitable. NAME, PHONE, and ADDRESS fall at 0.5-0.6 as they are personally identifying but more commonly shared.

You won't see any output yet. The next chunk adds the function that uses this config.

  • Add the classify_column function below the PII_PATTERNS dictionary in risk_scorer.py:
def classify_column(column_name):
    """Classify a column's PII type based on its name."""
    # Lowercase the column name for case-insensitive matching
    column_lower = column_name.lower()
    for pii_type, config in PII_PATTERNS.items():
        for pattern in config["column_patterns"]:
            # Check if the column name matches any PII pattern
            if re.search(pattern, column_lower):
                return {
                    "pii_type": pii_type,
                    "sensitivity": config["sensitivity"],
                    "tier": config["tier"],
                }
    # No PII pattern matched
    return {"pii_type": "NONE", "sensitivity": 0.0, "tier": "NONE"}

What does this function do?

The classify_column function takes a column name, lowercases it, and runs it against every regex pattern in PII_PATTERNS. The first match wins. If no pattern matches, the column is classified as NONE with zero sensitivity.

Using re.search instead of exact equality means a column named ssn_masked still matches the ssn pattern. Derived columns retain their PII classification.

  • Save risk_scorer.py.
  • Test the classifier by running this command in your terminal from the lineage-engine folder:
python -c "from risk_scorer import classify_column; print(classify_column('ssn')); print(classify_column('customer_id'))"
  • You should see two dictionaries. The first shows pii_type: SSN with sensitivity 1.0 and tier HIGH. The second shows pii_type: NONE with sensitivity 0.0.

Seeing a ModuleNotFoundError?

Make sure you are running the command from inside the lineage-engine folder where risk_scorer.py, lineage.py, and pipeline.py all live. Also check that you saved the file before running.

Help me debug a ModuleNotFoundError when importing risk_scorer

Build the risk dimension functions

Knowing that a column holds PII is only half the picture. A Social Security Number that feeds into 5 downstream columns is a bigger risk than one that stays in a single stage. The next three functions measure how far and how broadly a column's data spreads through the pipeline.

  • Add the compute_propagation_risk function below classify_column in risk_scorer.py:
def compute_propagation_risk(G, node_id):
    """Compute propagation risk: ratio of downstream columns to total columns."""
    # Find all nodes reachable from this node
    descendants = nx.descendants(G, node_id)
    total_nodes = G.number_of_nodes()
    if total_nodes <= 1:
        return 0.0
    # Ratio of downstream reach to total graph size
    return round(len(descendants) / (total_nodes - 1), 4)

What does this function do?

Propagation risk measures how far a column's data spreads. nx.descendants(G, node_id) returns every node reachable downstream from the given node. Dividing that count by the total number of other nodes gives a 0.0 to 1.0 ratio.

A source column like raw_customers.ssn that feeds downstream transforms will have a higher propagation risk than a terminal column with no descendants.

You won't see a change yet. The next chunk adds the remaining two risk dimension functions.

  • Add the compute_exposure_score and compute_composite_risk functions below compute_propagation_risk in risk_scorer.py:
def compute_exposure_score(G, node_id):
    """Compute exposure: ratio of downstream stages reached to total other stages."""
    descendants = nx.descendants(G, node_id)
    # Collect the distinct pipeline stages that downstream nodes belong to
    downstream_stages = set()
    for desc in descendants:
        downstream_stages.add(G.nodes[desc].get("stage", ""))

    all_stages = set(G.nodes[n].get("stage", "") for n in G.nodes())
    current_stage = G.nodes[node_id].get("stage", "")
    other_stages = all_stages - {current_stage}

    if not other_stages:
        return 0.0
    # Ratio of stages reached to total other stages
    return round(len(downstream_stages) / len(other_stages), 4)


def compute_composite_risk(sensitivity, propagation, exposure):
    """Compute weighted composite risk: 0.5 * sensitivity + 0.3 * propagation + 0.2 * exposure."""
    return round(0.5 * sensitivity + 0.3 * propagation + 0.2 * exposure, 4)

What do these functions do?

compute_exposure_score measures cross-stage exposure. It counts how many distinct pipeline stages the data reaches downstream and divides by the total number of other stages. A column that touches 4 out of 5 other stages has high exposure.

compute_composite_risk combines all three dimensions into a single 0.0 to 1.0 score using the formula 0.5 * sensitivity + 0.3 * propagation + 0.2 * exposure. Sensitivity is weighted highest because the inherent data classification (SSN vs. merchant name) is the primary risk driver.

  • Save risk_scorer.py.

You won't see a change yet. The next function ties everything together by scoring every column in the graph.

Score all columns and verify results

With the classifier and all three risk dimensions built, the final function iterates every node in the lineage graph, classifies it, computes each risk dimension, and produces a complete scored result for each column.

  • Add the score_all_columns function below compute_composite_risk in risk_scorer.py:
def score_all_columns(G):
    """Score every column in the lineage graph and return a list of result dicts."""
    # Get propagation depth for every node using topological sort
    depths = compute_all_depths(G)
    results = []

    for node_id in G.nodes():
        node_data = G.nodes[node_id]
        column_name = node_data.get("column", "")

        # Classify PII type and compute all three risk dimensions
        classification = classify_column(column_name)
        propagation = compute_propagation_risk(G, node_id)
        exposure = compute_exposure_score(G, node_id)
        composite = compute_composite_risk(
            classification["sensitivity"], propagation, exposure
        )
        # Count downstream nodes for blast radius
        descendants = nx.descendants(G, node_id)

        results.append({
            "node_id": node_id,
            "stage": node_data.get("stage", ""),
            "column": column_name,
            "stage_type": node_data.get("stage_type", ""),
            "pii_type": classification["pii_type"],
            "sensitivity_tier": classification["tier"],
            "sensitivity_score": classification["sensitivity"],
            "propagation_risk": propagation,
            "exposure_score": exposure,
            "composite_risk": composite,
            "propagation_depth": depths.get(node_id, 0),
            "blast_radius": len(descendants),
        })

    return results

What does this function do?

score_all_columns is the orchestrator. It first computes depth for every node using compute_all_depths from lineage.py, then loops through every node in the graph.

For each node, it classifies PII type, computes the three risk dimensions, combines them into a composite risk score, and counts the blast radius (total downstream descendants). Each result dictionary captures 12 fields that the report generator will use in the next step.

  • Save risk_scorer.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved risk_scorer.py before moving on to the test.

ⓧ I'd like to double check the full code

Here is the complete risk_scorer.py file. Compare your code against this:

"""PII detection and multi-dimensional risk scoring engine."""

import re

import networkx as nx

from lineage import compute_all_depths


PII_PATTERNS = {
    "SSN": {
        "column_patterns": [r"ssn", r"social_security"],
        "sensitivity": 1.0,
        "tier": "HIGH",
    },
    "EMAIL": {
        "column_patterns": [r"email", r"e_mail"],
        "sensitivity": 0.7,
        "tier": "HIGH",
    },
    "PHONE": {
        "column_patterns": [r"phone", r"mobile", r"telephone"],
        "sensitivity": 0.6,
        "tier": "MEDIUM",
    },
    "NAME": {
        "column_patterns": [r"first_name", r"last_name", r"full_name"],
        "sensitivity": 0.5,
        "tier": "MEDIUM",
    },
    "DOB": {
        "column_patterns": [r"date_of_birth", r"dob", r"birth_date"],
        "sensitivity": 0.7,
        "tier": "HIGH",
    },
    "ADDRESS": {
        "column_patterns": [r"address", r"street", r"zip_code"],
        "sensitivity": 0.6,
        "tier": "MEDIUM",
    },
    "CARD_NUMBER": {
        "column_patterns": [r"card_number", r"card_num", r"credit_card"],
        "sensitivity": 1.0,
        "tier": "HIGH",
    },
}


def classify_column(column_name):
    """Classify a column's PII type based on its name."""
    column_lower = column_name.lower()
    for pii_type, config in PII_PATTERNS.items():
        for pattern in config["column_patterns"]:
            if re.search(pattern, column_lower):
                return {
                    "pii_type": pii_type,
                    "sensitivity": config["sensitivity"],
                    "tier": config["tier"],
                }
    return {"pii_type": "NONE", "sensitivity": 0.0, "tier": "NONE"}


def compute_propagation_risk(G, node_id):
    """Compute propagation risk: ratio of downstream columns to total columns."""
    descendants = nx.descendants(G, node_id)
    total_nodes = G.number_of_nodes()
    if total_nodes <= 1:
        return 0.0
    return round(len(descendants) / (total_nodes - 1), 4)


def compute_exposure_score(G, node_id):
    """Compute exposure: ratio of downstream stages reached to total other stages."""
    descendants = nx.descendants(G, node_id)
    downstream_stages = set()
    for desc in descendants:
        downstream_stages.add(G.nodes[desc].get("stage", ""))

    all_stages = set(G.nodes[n].get("stage", "") for n in G.nodes())
    current_stage = G.nodes[node_id].get("stage", "")
    other_stages = all_stages - {current_stage}

    if not other_stages:
        return 0.0
    return round(len(downstream_stages) / len(other_stages), 4)


def compute_composite_risk(sensitivity, propagation, exposure):
    """Compute weighted composite risk: 0.5 * sensitivity + 0.3 * propagation + 0.2 * exposure."""
    return round(0.5 * sensitivity + 0.3 * propagation + 0.2 * exposure, 4)


def score_all_columns(G):
    """Score every column in the lineage graph and return a list of result dicts."""
    depths = compute_all_depths(G)
    results = []

    for node_id in G.nodes():
        node_data = G.nodes[node_id]
        column_name = node_data.get("column", "")

        classification = classify_column(column_name)
        propagation = compute_propagation_risk(G, node_id)
        exposure = compute_exposure_score(G, node_id)
        composite = compute_composite_risk(
            classification["sensitivity"], propagation, exposure
        )
        descendants = nx.descendants(G, node_id)

        results.append({
            "node_id": node_id,
            "stage": node_data.get("stage", ""),
            "column": column_name,
            "stage_type": node_data.get("stage_type", ""),
            "pii_type": classification["pii_type"],
            "sensitivity_tier": classification["tier"],
            "sensitivity_score": classification["sensitivity"],
            "propagation_risk": propagation,
            "exposure_score": exposure,
            "composite_risk": composite,
            "propagation_depth": depths.get(node_id, 0),
            "blast_radius": len(descendants),
        })

    return results
  • Verify the scoring engine by running this test command:
python -c "
from pipeline import generate_sample_data
from lineage import build_lineage_graph
from risk_scorer import score_all_columns

generate_sample_data()
G = build_lineage_graph()
results = score_all_columns(G)

print(f'Total scored entries: {len(results)}')

ssn = next(r for r in results if r['node_id'] == 'raw_customers.ssn')
print(f'SSN sensitivity: {ssn[\"sensitivity_score\"]} tier: {ssn[\"sensitivity_tier\"]}')

cid = next(r for r in results if r['node_id'] == 'raw_customers.customer_id')
print(f'Customer ID PII type: {cid[\"pii_type\"]}')

pii_count = sum(1 for r in results if r['pii_type'] != 'NONE')
print(f'Total PII columns: {pii_count}')
"
  • You should see four lines of output:
  • Total scored entries: 38 confirms every column in the graph was scored.
  • SSN sensitivity: 1.0 tier: HIGH confirms the SSN column received the highest sensitivity classification.
  • Customer ID PII type: NONE confirms non-PII columns are correctly classified.
  • Total PII columns: 16 confirms 16 out of 38 columns across all pipeline stages are detected as PII.

Why 16 PII columns and not just the 8 raw source columns?

PII columns appear in multiple pipeline stages. For example, email exists in raw_customers, cleaned_customers, and enriched_transactions. Each instance is tracked separately because it represents a distinct point in the pipeline where that PII can be accessed or leaked.

This is exactly why lineage-aware risk scoring matters. A naive count of source PII fields underestimates the true exposure surface.

Seeing different numbers or an error?

If you see a NameError or ImportError, check that all five functions (classify_column, compute_propagation_risk, compute_exposure_score, compute_composite_risk, score_all_columns) are present in risk_scorer.py and that your from lineage import compute_all_depths import line is at the top of the file.

If the PII count is different from 16, compare your PII_PATTERNS dictionary against the double-check tab above. A missing or misspelled regex pattern will change the detection count.

Help me debug my risk scorer output numbers

Every column in your pipeline now has a quantified risk profile. Next up, you will build the report generator that turns these scores into governance-ready Markdown and JSON deliverables, and wire up the CLI entry point to run the full engine end to end.

Generate Reports and Run the Engine

Your risk scoring engine can now classify every column for PII and compute a composite risk score. The numbers exist, but they are trapped in Python dictionaries.

In a real Data Office, stakeholders need those numbers in a structured report they can review and act on. In this step, you will build a report generator that turns scored data into governance-ready Markdown and JSON deliverables, then wire everything together with a CLI entry point.

In this step, get ready to:
  • Create a report generator that outputs a Markdown lineage report and a JSON risk summary.
  • Create a CLI entry point that orchestrates the full pipeline and prints a formatted risk dashboard.
  • Run the engine end-to-end and verify the console output and generated files.
Create the report generator

The report generator takes scored column data and graph metrics as inputs, then produces three outputs: a summary metrics dictionary, a Markdown report file, and a JSON export file.

  • Create a new file called report.py inside your lineage-engine folder.
  • Add the imports and the compute_summary_metrics function by copying this code into report.py:
"""Report generator - produces Markdown lineage report and JSON summary."""

import json
import os
from datetime import datetime


def compute_summary_metrics(scored_columns, graph_metrics):
    """Compute aggregate numerical metrics for the pipeline."""
    total = len(scored_columns)

    # Count columns in each sensitivity tier
    tier_counts = {"HIGH": 0, "MEDIUM": 0, "LOW": 0, "NONE": 0}
    for col in scored_columns:
        tier = col["sensitivity_tier"]
        tier_counts[tier] = tier_counts.get(tier, 0) + 1

    # PII detection rate across the pipeline
    classified = sum(1 for c in scored_columns if c["pii_type"] != "NONE")
    pii_rate = round(classified / total * 100, 1) if total > 0 else 0.0

    # Overall pipeline risk from composite scores
    avg_risk = round(
        sum(c["composite_risk"] for c in scored_columns) / total, 4
    ) if total > 0 else 0.0
    max_risk = max((c["composite_risk"] for c in scored_columns), default=0.0)

    # Blast radius stats for sensitive columns only
    sensitive_cols = [c for c in scored_columns if c["sensitivity_score"] > 0]
    blast_radii = [c["blast_radius"] for c in sensitive_cols]
    max_blast = max(blast_radii, default=0)
    avg_blast = round(sum(blast_radii) / len(blast_radii), 1) if blast_radii else 0.0

    # Critical paths: HIGH sensitivity columns with large blast radius
    critical = sum(
        1 for c in scored_columns
        if c["sensitivity_tier"] == "HIGH" and c["blast_radius"] > 3
    )

    max_depth = max((c["propagation_depth"] for c in scored_columns), default=0)

    return {
        "total_columns_tracked": total,
        "pii_columns_detected": classified,
        "pii_detection_rate_pct": pii_rate,
        "lineage_completeness_pct": 100.0,
        "pipeline_risk_score": avg_risk,
        "max_composite_risk": max_risk,
        "max_blast_radius": max_blast,
        "avg_blast_radius_sensitive": avg_blast,
        "critical_paths": critical,
        "max_propagation_depth": max_depth,
        "risk_distribution": tier_counts,
        "graph": graph_metrics,
    }

What does this code do?

The compute_summary_metrics function distills all 38 scored columns into a single summary dictionary with 10+ numerical metrics.

  • tier_counts tallies how many columns fall into each sensitivity tier (HIGH, MEDIUM, LOW, NONE).
  • pii_rate calculates the percentage of columns classified as PII across the entire pipeline.
  • critical counts HIGH-sensitivity columns with a blast radius greater than 3, identifying the most dangerous data paths.

You won't see output yet. The next two functions will use this summary to generate report files.

  • Add the generate_markdown_report and export_json_summary functions below compute_summary_metrics in report.py:

def generate_markdown_report(scored_columns, summary, output_path="output/lineage_report.md"):
    """Generate a Markdown lineage and risk report."""
    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    lines = []
    lines.append("# Data Lineage and Risk Scoring Report")
    lines.append(f"\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")

    # Table 1: Pipeline summary metrics
    lines.append("## Pipeline Summary Metrics\n")
    lines.append("| Metric | Value |")
    lines.append("|---|---|")
    lines.append(f"| Total Columns Tracked | {summary['total_columns_tracked']} |")
    lines.append(f"| PII Columns Detected | {summary['pii_columns_detected']} |")
    lines.append(f"| PII Detection Rate | {summary['pii_detection_rate_pct']}% |")
    lines.append(f"| Lineage Completeness | {summary['lineage_completeness_pct']}% |")
    lines.append(f"| Pipeline Risk Score | {summary['pipeline_risk_score']} |")
    lines.append(f"| Max Composite Risk | {summary['max_composite_risk']} |")
    lines.append(f"| Max Blast Radius | {summary['max_blast_radius']} columns |")
    lines.append(f"| Avg Blast Radius (Sensitive) | {summary['avg_blast_radius_sensitive']} columns |")
    lines.append(f"| Critical Paths | {summary['critical_paths']} |")
    lines.append(f"| Max Propagation Depth | {summary['max_propagation_depth']} stages |")
    lines.append("")

    # Table 2: Risk distribution by tier
    lines.append("## Risk Distribution\n")
    dist = summary["risk_distribution"]
    total = summary["total_columns_tracked"]
    lines.append("| Tier | Count | Percentage |")
    lines.append("|---|---|---|")
    for tier in ["HIGH", "MEDIUM", "LOW", "NONE"]:
        count = dist.get(tier, 0)
        pct = round(count / total * 100, 1) if total > 0 else 0.0
        lines.append(f"| {tier} | {count} | {pct}% |")
    lines.append("")

    # Table 3: Graph structure metrics
    lines.append("## Graph Metrics\n")
    gm = summary["graph"]
    lines.append("| Metric | Value |")
    lines.append("|---|---|")
    lines.append(f"| Total Nodes | {gm['total_nodes']} |")
    lines.append(f"| Total Edges | {gm['total_edges']} |")
    lines.append(f"| Source Columns | {gm['source_columns']} |")
    lines.append(f"| Terminal Columns | {gm['terminal_columns']} |")
    lines.append(f"| Avg In-Degree | {gm['avg_in_degree']} |")
    lines.append(f"| Avg Out-Degree | {gm['avg_out_degree']} |")
    lines.append(f"| Longest Path | {gm['longest_path_length']} stages |")
    lines.append("")

    # Table 4: Detailed breakdown of HIGH-tier columns
    lines.append("## High-Risk Column Details\n")
    high_risk = sorted(
        [c for c in scored_columns if c["sensitivity_tier"] == "HIGH"],
        key=lambda x: x["composite_risk"],
        reverse=True,
    )
    if high_risk:
        lines.append("| Column | Stage | PII Type | Sensitivity | Propagation | Composite | Blast Radius |")
        lines.append("|---|---|---|---|---|---|---|")
        for col in high_risk:
            lines.append(
                f"| {col['column']} | {col['stage']} | {col['pii_type']} | "
                f"{col['sensitivity_score']} | {col['propagation_risk']} | "
                f"{col['composite_risk']} | {col['blast_radius']} |"
            )
    else:
        lines.append("No high-risk columns detected.")
    lines.append("")

    # Table 5: Every column in the graph sorted by risk
    lines.append("## Full Column Inventory\n")
    lines.append("| Node | Type | PII | Tier | Composite Risk | Blast Radius | Depth |")
    lines.append("|---|---|---|---|---|---|---|")
    for col in sorted(scored_columns, key=lambda x: x["composite_risk"], reverse=True):
        lines.append(
            f"| {col['node_id']} | {col['stage_type']} | {col['pii_type']} | "
            f"{col['sensitivity_tier']} | {col['composite_risk']} | "
            f"{col['blast_radius']} | {col['propagation_depth']} |"
        )
    lines.append("")

    # Longest path through the lineage DAG
    lines.append("## Longest Lineage Path\n")
    path = summary["graph"]["longest_path"]
    if path:
        lines.append(" -> ".join(path))
    else:
        lines.append("No paths found.")
    lines.append("")

    report_text = "\n".join(lines)
    with open(output_path, "w") as f:
        f.write(report_text)

    return output_path


def export_json_summary(summary, scored_columns, output_path="output/risk_summary.json"):
    """Export the complete risk analysis as JSON."""
    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    # Bundle summary metrics and per-column scores for programmatic consumption
    export_data = {
        "generated_at": datetime.now().isoformat(),
        "summary": summary,
        "columns": scored_columns,
    }

    with open(output_path, "w") as f:
        json.dump(export_data, f, indent=2)

    return output_path

What do these functions do?

generate_markdown_report builds a structured Markdown file with 5 tables: Pipeline Summary Metrics, Risk Distribution, Graph Metrics, High-Risk Column Details, and Full Column Inventory. It also includes the longest lineage path through the DAG.

export_json_summary writes the entire analysis as a JSON file with summary and columns keys. This gives downstream tools or dashboards a machine-readable format to consume.

  • Save report.py.

You won't see a change yet. The next file, main.py, will call all three functions and produce visible output.

✔️ Awesome, I've got everything!

Great. Double check you have saved report.py before moving on.

ⓧ I'd like to double check the full code

"""Report generator - produces Markdown lineage report and JSON summary."""

import json
import os
from datetime import datetime


def compute_summary_metrics(scored_columns, graph_metrics):
    """Compute aggregate numerical metrics for the pipeline."""
    total = len(scored_columns)

    tier_counts = {"HIGH": 0, "MEDIUM": 0, "LOW": 0, "NONE": 0}
    for col in scored_columns:
        tier = col["sensitivity_tier"]
        tier_counts[tier] = tier_counts.get(tier, 0) + 1

    classified = sum(1 for c in scored_columns if c["pii_type"] != "NONE")
    pii_rate = round(classified / total * 100, 1) if total > 0 else 0.0

    avg_risk = round(
        sum(c["composite_risk"] for c in scored_columns) / total, 4
    ) if total > 0 else 0.0
    max_risk = max((c["composite_risk"] for c in scored_columns), default=0.0)

    sensitive_cols = [c for c in scored_columns if c["sensitivity_score"] > 0]
    blast_radii = [c["blast_radius"] for c in sensitive_cols]
    max_blast = max(blast_radii, default=0)
    avg_blast = round(sum(blast_radii) / len(blast_radii), 1) if blast_radii else 0.0

    critical = sum(
        1 for c in scored_columns
        if c["sensitivity_tier"] == "HIGH" and c["blast_radius"] > 3
    )

    max_depth = max((c["propagation_depth"] for c in scored_columns), default=0)

    return {
        "total_columns_tracked": total,
        "pii_columns_detected": classified,
        "pii_detection_rate_pct": pii_rate,
        "lineage_completeness_pct": 100.0,
        "pipeline_risk_score": avg_risk,
        "max_composite_risk": max_risk,
        "max_blast_radius": max_blast,
        "avg_blast_radius_sensitive": avg_blast,
        "critical_paths": critical,
        "max_propagation_depth": max_depth,
        "risk_distribution": tier_counts,
        "graph": graph_metrics,
    }


def generate_markdown_report(scored_columns, summary, output_path="output/lineage_report.md"):
    """Generate a Markdown lineage and risk report."""
    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    lines = []
    lines.append("# Data Lineage and Risk Scoring Report")
    lines.append(f"\nGenerated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")

    lines.append("## Pipeline Summary Metrics\n")
    lines.append("| Metric | Value |")
    lines.append("|---|---|")
    lines.append(f"| Total Columns Tracked | {summary['total_columns_tracked']} |")
    lines.append(f"| PII Columns Detected | {summary['pii_columns_detected']} |")
    lines.append(f"| PII Detection Rate | {summary['pii_detection_rate_pct']}% |")
    lines.append(f"| Lineage Completeness | {summary['lineage_completeness_pct']}% |")
    lines.append(f"| Pipeline Risk Score | {summary['pipeline_risk_score']} |")
    lines.append(f"| Max Composite Risk | {summary['max_composite_risk']} |")
    lines.append(f"| Max Blast Radius | {summary['max_blast_radius']} columns |")
    lines.append(f"| Avg Blast Radius (Sensitive) | {summary['avg_blast_radius_sensitive']} columns |")
    lines.append(f"| Critical Paths | {summary['critical_paths']} |")
    lines.append(f"| Max Propagation Depth | {summary['max_propagation_depth']} stages |")
    lines.append("")

    lines.append("## Risk Distribution\n")
    dist = summary["risk_distribution"]
    total = summary["total_columns_tracked"]
    lines.append("| Tier | Count | Percentage |")
    lines.append("|---|---|---|")
    for tier in ["HIGH", "MEDIUM", "LOW", "NONE"]:
        count = dist.get(tier, 0)
        pct = round(count / total * 100, 1) if total > 0 else 0.0
        lines.append(f"| {tier} | {count} | {pct}% |")
    lines.append("")

    lines.append("## Graph Metrics\n")
    gm = summary["graph"]
    lines.append("| Metric | Value |")
    lines.append("|---|---|")
    lines.append(f"| Total Nodes | {gm['total_nodes']} |")
    lines.append(f"| Total Edges | {gm['total_edges']} |")
    lines.append(f"| Source Columns | {gm['source_columns']} |")
    lines.append(f"| Terminal Columns | {gm['terminal_columns']} |")
    lines.append(f"| Avg In-Degree | {gm['avg_in_degree']} |")
    lines.append(f"| Avg Out-Degree | {gm['avg_out_degree']} |")
    lines.append(f"| Longest Path | {gm['longest_path_length']} stages |")
    lines.append("")

    lines.append("## High-Risk Column Details\n")
    high_risk = sorted(
        [c for c in scored_columns if c["sensitivity_tier"] == "HIGH"],
        key=lambda x: x["composite_risk"],
        reverse=True,
    )
    if high_risk:
        lines.append("| Column | Stage | PII Type | Sensitivity | Propagation | Composite | Blast Radius |")
        lines.append("|---|---|---|---|---|---|---|")
        for col in high_risk:
            lines.append(
                f"| {col['column']} | {col['stage']} | {col['pii_type']} | "
                f"{col['sensitivity_score']} | {col['propagation_risk']} | "
                f"{col['composite_risk']} | {col['blast_radius']} |"
            )
    else:
        lines.append("No high-risk columns detected.")
    lines.append("")

    lines.append("## Full Column Inventory\n")
    lines.append("| Node | Type | PII | Tier | Composite Risk | Blast Radius | Depth |")
    lines.append("|---|---|---|---|---|---|---|")
    for col in sorted(scored_columns, key=lambda x: x["composite_risk"], reverse=True):
        lines.append(
            f"| {col['node_id']} | {col['stage_type']} | {col['pii_type']} | "
            f"{col['sensitivity_tier']} | {col['composite_risk']} | "
            f"{col['blast_radius']} | {col['propagation_depth']} |"
        )
    lines.append("")

    lines.append("## Longest Lineage Path\n")
    path = summary["graph"]["longest_path"]
    if path:
        lines.append(" -> ".join(path))
    else:
        lines.append("No paths found.")
    lines.append("")

    report_text = "\n".join(lines)
    with open(output_path, "w") as f:
        f.write(report_text)

    return output_path


def export_json_summary(summary, scored_columns, output_path="output/risk_summary.json"):
    """Export the complete risk analysis as JSON."""
    os.makedirs(os.path.dirname(output_path), exist_ok=True)

    export_data = {
        "generated_at": datetime.now().isoformat(),
        "summary": summary,
        "columns": scored_columns,
    }

    with open(output_path, "w") as f:
        json.dump(export_data, f, indent=2)

    return output_path
Create the CLI entry point

The entry point ties all four modules together into a single command. It generates sample data, builds the lineage graph, scores every column, produces both report files, and prints a formatted dashboard to the console.

  • Create a new file called main.py inside your lineage-engine folder and paste this code:
"""Data Lineage and Risk Scoring Engine - entry point."""

from pipeline import generate_sample_data
from lineage import build_lineage_graph, get_graph_metrics
from risk_scorer import score_all_columns
from report import compute_summary_metrics, generate_markdown_report, export_json_summary


def main():
    print("=" * 60)
    print("  Data Lineage and Risk Scoring Engine")
    print("=" * 60)

    # Step 1: Generate CSV source data
    print("\n[1/4] Generating sample pipeline data...")
    datasets = generate_sample_data()
    for name, df in datasets.items():
        print(f"  {name}: {len(df)} rows, {len(df.columns)} columns")

    # Step 2: Build the column-level lineage DAG
    print("\n[2/4] Building column-level lineage graph...")
    G = build_lineage_graph()
    graph_metrics = get_graph_metrics(G)
    print(f"  Nodes: {graph_metrics['total_nodes']}")
    print(f"  Edges: {graph_metrics['total_edges']}")
    print(f"  Longest path: {graph_metrics['longest_path_length']} stages")

    # Step 3: Classify PII and compute risk scores
    print("\n[3/4] Running PII detection and risk scoring...")
    scored_columns = score_all_columns(G)
    high = [c for c in scored_columns if c["sensitivity_tier"] == "HIGH"]
    medium = [c for c in scored_columns if c["sensitivity_tier"] == "MEDIUM"]
    print(f"  HIGH risk: {len(high)} columns")
    print(f"  MEDIUM risk: {len(medium)} columns")

    # Step 4: Generate Markdown and JSON reports
    print("\n[4/4] Generating reports...")
    summary = compute_summary_metrics(scored_columns, graph_metrics)
    md_path = generate_markdown_report(scored_columns, summary)
    json_path = export_json_summary(summary, scored_columns)
    print(f"  Markdown: {md_path}")
    print(f"  JSON: {json_path}")

    # Print the formatted risk summary dashboard
    print("\n" + "=" * 60)
    print("  PIPELINE RISK SUMMARY")
    print("=" * 60)
    print(f"  Columns Tracked:        {summary['total_columns_tracked']}")
    print(f"  PII Detection Rate:     {summary['pii_detection_rate_pct']}%")
    print(f"  Lineage Completeness:   {summary['lineage_completeness_pct']}%")
    print(f"  Pipeline Risk Score:    {summary['pipeline_risk_score']}")
    print(f"  Max Composite Risk:     {summary['max_composite_risk']}")
    print(f"  Max Blast Radius:       {summary['max_blast_radius']} columns")
    print(f"  Critical Paths:         {summary['critical_paths']}")
    print(f"  Max Propagation Depth:  {summary['max_propagation_depth']} stages")
    print("")

    # Visual risk distribution bar chart
    dist = summary["risk_distribution"]
    total = summary["total_columns_tracked"]
    print("  Risk Distribution:")
    for tier in ["HIGH", "MEDIUM", "LOW", "NONE"]:
        count = dist.get(tier, 0)
        pct = round(count / total * 100, 1) if total > 0 else 0.0
        bar = "#" * int(pct / 2)
        print(f"    {tier:6s}: {count:3d} ({pct:5.1f}%) {bar}")

    print("\n" + "=" * 60)
    print("  Reports saved to output/")
    print("=" * 60)


if __name__ == "__main__":
    main()

What does this code do?

The main() function orchestrates all four modules in sequence. It prints progress for each of the 4 phases so you can see the pipeline running step by step.

  • Phases [1/4] through [4/4] call generate_sample_data, build_lineage_graph, score_all_columns, and the three report functions.
  • The PIPELINE RISK SUMMARY block prints all key numerical metrics: Pipeline Risk Score, PII Detection Rate, Max Blast Radius, Critical Paths, and Max Propagation Depth.
  • The Risk Distribution section uses # characters to create a visual bar chart in the console, showing the proportion of columns in each sensitivity tier.
  • Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before running the engine.

ⓧ I'd like to double check the full code

"""Data Lineage and Risk Scoring Engine - entry point."""

from pipeline import generate_sample_data
from lineage import build_lineage_graph, get_graph_metrics
from risk_scorer import score_all_columns
from report import compute_summary_metrics, generate_markdown_report, export_json_summary


def main():
    print("=" * 60)
    print("  Data Lineage and Risk Scoring Engine")
    print("=" * 60)

    print("\n[1/4] Generating sample pipeline data...")
    datasets = generate_sample_data()
    for name, df in datasets.items():
        print(f"  {name}: {len(df)} rows, {len(df.columns)} columns")

    print("\n[2/4] Building column-level lineage graph...")
    G = build_lineage_graph()
    graph_metrics = get_graph_metrics(G)
    print(f"  Nodes: {graph_metrics['total_nodes']}")
    print(f"  Edges: {graph_metrics['total_edges']}")
    print(f"  Longest path: {graph_metrics['longest_path_length']} stages")

    print("\n[3/4] Running PII detection and risk scoring...")
    scored_columns = score_all_columns(G)
    high = [c for c in scored_columns if c["sensitivity_tier"] == "HIGH"]
    medium = [c for c in scored_columns if c["sensitivity_tier"] == "MEDIUM"]
    print(f"  HIGH risk: {len(high)} columns")
    print(f"  MEDIUM risk: {len(medium)} columns")

    print("\n[4/4] Generating reports...")
    summary = compute_summary_metrics(scored_columns, graph_metrics)
    md_path = generate_markdown_report(scored_columns, summary)
    json_path = export_json_summary(summary, scored_columns)
    print(f"  Markdown: {md_path}")
    print(f"  JSON: {json_path}")

    print("\n" + "=" * 60)
    print("  PIPELINE RISK SUMMARY")
    print("=" * 60)
    print(f"  Columns Tracked:        {summary['total_columns_tracked']}")
    print(f"  PII Detection Rate:     {summary['pii_detection_rate_pct']}%")
    print(f"  Lineage Completeness:   {summary['lineage_completeness_pct']}%")
    print(f"  Pipeline Risk Score:    {summary['pipeline_risk_score']}")
    print(f"  Max Composite Risk:     {summary['max_composite_risk']}")
    print(f"  Max Blast Radius:       {summary['max_blast_radius']} columns")
    print(f"  Critical Paths:         {summary['critical_paths']}")
    print(f"  Max Propagation Depth:  {summary['max_propagation_depth']} stages")
    print("")

    dist = summary["risk_distribution"]
    total = summary["total_columns_tracked"]
    print("  Risk Distribution:")
    for tier in ["HIGH", "MEDIUM", "LOW", "NONE"]:
        count = dist.get(tier, 0)
        pct = round(count / total * 100, 1) if total > 0 else 0.0
        bar = "#" * int(pct / 2)
        print(f"    {tier:6s}: {count:3d} ({pct:5.1f}%) {bar}")

    print("\n" + "=" * 60)
    print("  Reports saved to output/")
    print("=" * 60)


if __name__ == "__main__":
    main()
Run the engine and review the outputs

Everything is wired up. Time to run the full pipeline end-to-end and see the results.

  • Make sure you are inside the lineage-engine folder in your terminal.
  • Run the engine by executing this command:
python main.py

You should see a full console dashboard with 4 progress phases followed by a PIPELINE RISK SUMMARY section. The summary includes Pipeline Risk Score, PII Detection Rate, Max Blast Radius, Critical Paths, Max Propagation Depth, and a visual Risk Distribution bar chart.

  • Confirm the report files were created by running:
ls output/

You should see lineage_report.md and risk_summary.json in the output.

  • Open output/lineage_report.md in your editor to review the Markdown report.

The report contains 5 tables: Pipeline Summary Metrics, Risk Distribution, Graph Metrics, High-Risk Column Details, and Full Column Inventory. It also includes a Longest Lineage Path section.

  • Open output/risk_summary.json in your editor.

The JSON file contains two top-level keys: summary (the 10+ metrics dictionary) and columns (all 38 scored column entries).

Seeing a ModuleNotFoundError?

  • Make sure your terminal is inside the lineage-engine folder where all five .py files live.
  • Double check that pipeline.py, lineage.py, risk_scorer.py, and report.py are all saved with the exact filenames shown.
  • Verify that networkx and pandas are installed in the same Python environment you are running the script with.

Help me debug a ModuleNotFoundError when running python main.py for my data lineage project.

Why both Markdown and JSON?

The Markdown report is for human stakeholders. A Data Governance lead can open it, scan the tables, and immediately see which columns are high risk and how far PII spreads.

The JSON export is for downstream tooling. Other systems (dashboards, alerting, CI pipelines) can parse the file and act on the numbers programmatically.

Secret mission

Add Interactive Impact Queries and DOT Graph Export

Your engine scores risk and generates reports. But stakeholders also need visual lineage maps they can present in meetings, and analysts need on-demand impact answers without re-running the full pipeline. In this mission, you will export the lineage graph as a color-coded DOT file and build an interactive CLI for querying any column's upstream and downstream blast radius.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs.

Resources you used:

  • The lineage-engine project folder containing 5 Python modules, sample CSV data in data/, and generated reports in output/.
  • The networkx and pandas Python packages installed via pip.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with the lineage engine or want to extend it further.

  • Your project files stay in the lineage-engine folder on your Desktop.
  • The networkx and pandas packages remain installed in your Python environment.
  • There are zero ongoing costs since everything is local.

Pause - I'll come back to this later

Since this project is entirely local, there are no running processes to pause. Your files and packages will be exactly where you left them.

  • Close your code editor and any open terminal windows.
  • When you come back, open a terminal, navigate to the lineage-engine folder on your Desktop, and run the engine again:
cd ~/Desktop/lineage-engine
python main.py

Delete - I don't want to use this again

Remove all project files and uninstall the Python packages to clean up completely.

  • Delete the project folder by running the command for your operating system.

macOS

cd ~/Desktop
rm -rf lineage-engine

Windows

cd ~/Desktop
Remove-Item -Recurse -Force lineage-engine
  • Uninstall the Python packages by running:
pip uninstall networkx pandas
  • Type y and press Enter when prompted to confirm each package removal.

That's a Wrap!

That's a Wrap!

Nice work! You just built a fully functional data lineage and risk scoring engine from scratch.

This is the exact kind of quantified risk reporting a Data Office presents to stakeholders. You now know how to trace every column through a pipeline, score it for risk, and generate the documentation to prove it.

You've learned how to:

  • Build a column-level lineage DAG (38 nodes, 26 edges) that models a 6-stage data pipeline using NetworkX DiGraph and topological sort for depth computation.
  • Implement a multi-dimensional risk scoring engine with regex-based PII detection across 7 PII types and a weighted composite formula combining sensitivity, propagation, and exposure scores.
  • Generate governance-ready Markdown and JSON reports with 10+ numerical metrics including pipeline risk score, PII detection rate, blast radius, and critical paths.
  • Secret Mission: Add interactive impact queries for on-demand column analysis and a DOT graph export with risk-tier color coding for visual lineage documentation.

Ready to quiz yourself?