Come implementare il data lineage: guida passo-passo

Esempio: script Python per il lineage Snowflake

import snowflake.connector
import json
from collections import defaultdict

conn = snowflake.connector.connect(
    account='YOUR_ACCOUNT',
    user='YOUR_USER',
    password='YOUR_PASSWORD',
    warehouse='COMPUTE_WH',
    database='SNOWFLAKE',
    schema='ACCOUNT_USAGE'
)

query = """
SELECT 
    query_id,
    query_text,
    database_name,
    schema_name,
    user_name,
    start_time
FROM snowflake.account_usage.query_history
WHERE query_type IN ('INSERT', 'MERGE', 'CREATE_TABLE_AS_SELECT')
  AND start_time >= DATEADD(day, -7, CURRENT_TIMESTAMP())
ORDER BY start_time DESC
LIMIT 1000;
"""

cursor = conn.cursor()
cursor.execute(query)

lineage_graph = defaultdict(list)

for row in cursor:
    query_id, query_text, db, schema, user, timestamp = row
    
    if 'INSERT INTO' in query_text.upper():
        target = extract_table_name(query_text, 'INSERT INTO')
        sources = extract_table_names(query_text, ['FROM', 'JOIN'])
        
        for source in sources:
            lineage_graph[source].append({
                'target': target,
                'query_id': query_id,
                'user': user,
                'timestamp': str(timestamp)
            })

with open('lineage_output.json', 'w') as f:
    json.dump(dict(lineage_graph), f, indent=2)

print(f"Estratto lineage per {len(lineage_graph)} tabelle sorgente")
conn.close()

Esempio: DAG Airflow per estrarre il lineage

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests

def extract_snowflake_lineage():
    """Estrai il lineage da Snowflake e invia al catalogo"""
    lineage_data = get_snowflake_lineage()
    
    response = requests.post(
        'https://catalog.example.com/api/lineage',
        headers={'Authorization': f'Bearer {CATALOG_API_KEY}'},
        json=lineage_data
    )
    response.raise_for_status()
    print(f"Caricate {len(lineage_data)} relazioni di lineage")

def extract_dbt_lineage():
    """Parsa il manifest dbt e carica"""
    with open('/dbt/target/manifest.json') as f:
        manifest = json.load(f)
    
    response = requests.post(
        'https://catalog.example.com/api/dbt/manifest',
        headers={'Authorization': f'Bearer {CATALOG_API_KEY}'},
        json=manifest
    )
    response.raise_for_status()

default_args = {
    'owner': 'data-platform',
    'depends_on_past': False,
    'start_date': datetime(2025, 1, 1),
    'email_on_failure': True,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'lineage_extraction',
    default_args=default_args,
    description='Estrazione giornaliera metadati di lineage',
    schedule_interval='@daily',
    catchup=False,
) as dag:

    extract_snowflake = PythonOperator(
        task_id='extract_snowflake_lineage',
        python_callable=extract_snowflake_lineage,
    )

    extract_dbt = PythonOperator(
        task_id='extract_dbt_lineage',
        python_callable=extract_dbt_lineage,
    )

    extract_snowflake >> extract_dbt

Come implementare il data lineage?

1) Scope + obiettivi, 2) strumenti (catalogo, open source, manuale), 3) architettura (storage, ingestion, visualizzazione), 4) cattura auto (log, API, manifest), 5) integrazione nello stack (warehouse, ETL, BI), 6) validazione, 7) governance, 8) formazione. Tempo: 1-3 mesi con tool moderni, 3-6 mesi con piattaforma enterprise.

Qual è il modo migliore per automatizzare?

1) Parsing log SQL (Snowflake, BigQuery, Redshift), 2) API metadati (Looker, Tableau), 3) manifest dbt, 4) OpenLineage per Airflow/Spark. Atlan o Select Star automatizzano questi metodi.

Quanto tempo serve?

Cataloghi cloud: 1-4 settimane. Piattaforme enterprise: 2-6 mesi. Open source: 1-3 mesi. Strumenti visuali (Datadef): immediato ma manutenzione manuale.

Quali strumenti servono?

1) Piattaforma lineage (Atlan, Collibra, DataHub), 2) estrattori per il tuo stack (dbt, Airflow, Looker/Tableau), 3) accesso ai log SQL (ACCOUNT_USAGE, INFORMATION_SCHEMA), 4) orchestrazione per refresh, 5) opzione: OpenLineage e documentazione visuale (Datadef).