Como implementar data lineage: guia passo a passo

Exemplo: script Python para lineage no 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"Lineage extraído para {len(lineage_graph)} tabelas fonte")
conn.close()

Exemplo: DAG Airflow para extrair lineage

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

def extract_snowflake_lineage():
    """Extrai lineage do Snowflake e envia para o catálogo"""
    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"Enviadas {len(lineage_data)} arestas de lineage")

def extract_dbt_lineage():
    """Parsa o manifest do dbt e envia"""
    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='Extração diária de metadados de 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

Como implementar data lineage?

1) Escopo + objetivos, 2) ferramentas (catálogo, open source, manual), 3) arquitetura (storage, ingestion, visualização), 4) captura automática (logs, APIs, manifests), 5) integração no stack (warehouse, ETL, BI), 6) validação, 7) governança, 8) treinamento. Prazo: 1-3 meses com ferramentas modernas, 3-6 meses com plataforma enterprise.

Qual o melhor jeito de automatizar?

1) Parsing de logs SQL (Snowflake, BigQuery, Redshift), 2) APIs de metadados (Looker, Tableau), 3) manifests do dbt, 4) OpenLineage para Airflow/Spark. Atlan ou Select Star automatizam esses métodos.

Quanto tempo leva?

Catálogos cloud: 1-4 semanas. Plataformas enterprise: 2-6 meses. Open source: 1-3 meses. Ferramentas visuais (Datadef): imediato mas manutenção manual.

Quais ferramentas são necessárias?

1) Plataforma de lineage (Atlan, Collibra, DataHub), 2) extratores para seu stack (dbt, Airflow, Looker/Tableau), 3) acesso aos logs SQL (ACCOUNT_USAGE, INFORMATION_SCHEMA), 4) orquestração para refresh, 5) opcional: OpenLineage e documentação visual (Datadef).