Comment implémenter le data lineage : guide pas à pas

Exemple : script Python pour le lineage Snowflake

import snowflake.connector
import json
from collections import defaultdict

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

# Query recent queries for lineage
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)

# Parse queries and build lineage
lineage_graph = defaultdict(list)

for row in cursor:
    query_id, query_text, db, schema, user, timestamp = row
    
    # Simple parsing (use sqlparse or sqlglot for production)
    if 'INSERT INTO' in query_text.upper():
        # Extract target table
        target = extract_table_name(query_text, 'INSERT INTO')
        # Extract source tables from FROM/JOIN clauses
        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)
            })

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

print(f"Extracted lineage for {len(lineage_graph)} source tables")
conn.close()

Exemple : DAG Airflow pour extraire le lineage

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

def extract_snowflake_lineage():
    """Extract lineage from Snowflake and send to catalog"""
    # Your lineage extraction logic here
    lineage_data = get_snowflake_lineage()
    
    # Upload to catalog tool API
    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"Uploaded {len(lineage_data)} lineage edges")

def extract_dbt_lineage():
    """Parse dbt manifest and upload"""
    with open('/dbt/target/manifest.json') as f:
        manifest = json.load(f)
    
    # Upload to catalog
    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='Daily lineage metadata extraction',
    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

Comment implémenter le data lineage ?

1) Scope + objectifs, 2) outils (catalogue, open source, manuel), 3) architecture (stockage, ingestion, visualisation), 4) capture auto (logs, APIs, manifestes), 5) intégration au stack (warehouse, ETL, BI), 6) validation, 7) gouvernance, 8) formation. Délai : 1-3 mois avec outils modernes, 3-6 mois sur plateforme enterprise.

Quelle est la meilleure façon d’automatiser ?

1) Parsing des logs SQL (Snowflake, BigQuery, Redshift), 2) APIs métadonnées (Looker, Tableau), 3) manifestes dbt, 4) OpenLineage pour Airflow/Spark. Atlan ou Select Star automatisent ces méthodes.

Combien de temps cela prend ?

Catalogues cloud : 1-4 semaines. Plateformes enterprise : 2-6 mois. Open source : 1-3 mois. Outils visuels (Datadef) : immédiat mais maintenance manuelle.

Quels outils sont nécessaires ?

1) Plateforme lineage (Atlan, Collibra, DataHub), 2) extracteurs pour ton stack (dbt, Airflow, Looker/Tableau), 3) accès logs SQL (ACCOUNT_USAGE, INFORMATION_SCHEMA), 4) orchestration pour rafraîchir, 5) option : OpenLineage et doc visuelle (Datadef).