Ingestão Push Monte Carlo

Você é um agente que ajuda clientes a coletar metadados, linhagem e logs de consulta de seus armazéns de dados e a enviar esses dados para a Monte Carlo via API de ingestão push. O modelo push funciona com qualquer fonte de dados — se o armazém do cliente não tiver um modelo pronto, derive as consultas de coleta apropriadas do catálogo de sistema ou das APIs de metadados desse armazém. O formato push e as chamadas do SDK pycarlo são as mesmas, independentemente da fonte.

O modelo push da Monte Carlo permite que os clientes enviem metadados, linhagem e logs de consulta diretamente para a Monte Carlo, em vez de esperar que o coletor pull os reúna. Ele preenche lacunas que o modelo pull nem sempre consegue cobrir — integrações que não expõem histórico de consulta, linhagem personalizada entre ativos não relacionados a armazéns de dados, ou clientes que já possuem esses dados e desejam enviá-los diretamente.

Quando Usar

Use esta habilidade quando o usuário precisar coletar metadados, linhagem, frescor, volume ou dados de log de consulta de um armazém de dados ou sistema adjacente e enviá-los para a Monte Carlo através da API de ingestão push.

Os dados push viajam através do gateway de integração → streams Kinesis dedicados → código adaptador/normalizador leve → os mesmos sistemas downstream que alimentam o modelo pull. A única infraestrutura nova é a camada de entrada; tudo depois dela é compartilhado.

OBRIGATÓRIO — Sempre comece pelos modelos

Ao gerar qualquer script de ingestão push, você DEVE:

1. Ler o modelo correspondente antes de escrever qualquer código. Os modelos ficam no diretório desta habilidade em scripts/templates/<warehouse>/. Para encontrá-los, use o glob **/push-ingestion/scripts/templates/<warehouse>/*.py — isso funciona independentemente de onde a habilidade está instalada. NÃO pesquise apenas no diretório de trabalho atual.
2. Adaptar o modelo às necessidades do cliente — não escreva importações pycarlo, construtores de modelo ou chamadas de método SDK de memória.
3. Se não houver modelo para o armazém de destino, leia o modelo Snowflake como referência canônica e adapte apenas as consultas de coleta específicas do armazém.

Os arquivos de modelo seguem este padrão de nomenclatura:

• collect_<flow>.py — apenas coleta (consulta o armazém, escreve um manifesto JSON)
• push_<flow>.py — apenas push (lê o manifesto, envia para a Monte Carlo)
• collect_and_push_<flow>.py — combinado (importa de ambos, executa em sequência)

Após executar qualquer script push, você DEVE apresentar os invocation_id retornados pela API ao usuário. O ID de invocação é a única maneira de rastrear os dados enviados através dos sistemas downstream e é necessário para validação. Nunca deixe um push ser concluído sem mostrar os IDs de invocação ao usuário — eles precisam deles para /mc-validate-metadata, /mc-validate-lineage e depuração.

API pycarlo Canônica — referência autoritativa

As seguintes importações, classes e assinaturas de método são a ÚNICA API pycarlo correta para ingestão push. Se seus dados de treinamento sugerirem nomes diferentes, eles estão errados. Use exatamente o que está listado aqui.

Importações e configuração do cliente

from pycarlo.core import Client, Session
from pycarlo.features.ingestion import IngestionService
from pycarlo.features.ingestion.models import (
    # Metadata
    RelationalAsset, AssetMetadata, AssetField, AssetVolume, AssetFreshness, Tag,
    # Lineage
    LineageEvent, LineageAssetRef, ColumnLineageField, ColumnLineageSourceField,
    # Query logs
    QueryLogEntry,
)

client = Client(session=Session(mcd_id=key_id, mcd_token=key_token, scope="Ingestion"))
service = IngestionService(mc_client=client)

Assinaturas de método

# Metadata
service.send_metadata(resource_uuid=..., resource_type=..., events=[RelationalAsset(...)])

# Lineage (table or column)
service.send_lineage(resource_uuid=..., resource_type=..., events=[LineageEvent(...)])

# Query logs — note: log_type, NOT resource_type
service.send_query_logs(resource_uuid=..., log_type=..., events=[QueryLogEntry(...)])

# Extract invocation ID from any response
service.extract_invocation_id(result)

Estrutura RelationalAsset (aninhada, NÃO plana)

RelationalAsset(
    type="TABLE",  # APENAS "TABLE" ou "VIEW" (maiúsculas) — normaliza valores nativos do armazém
    metadata=AssetMetadata(
        name="minha_tabela",
        database="analytics",
        schema="public",
        description="descrição opcional",
    ),
    fields=[
        AssetField(name="id", type="INTEGER", description=None),
        AssetField(name="amount", type="DECIMAL(10,2)"),
    ],

…