Contrato de ingestão.

O contrato de ingestão é a definição executável de um pipeline de tabela: origem, destino, modo de gravação, política de esquema, regras de qualidade, transformações, controles de execução e comportamento de runtime.

Exemplo completo

YAML

_metadata:
  contract_version: 1.0.0

source:
  type: connector
  connector: postgres
  options:
    url: "{{ secret:crm/postgres_url }}"
    dbtable: public.customers
    driver: org.postgresql.Driver
  read:
    partition_column: customer_id
    lower_bound: 1
    upper_bound: 1000000
    num_partitions: 8
    fetchsize: 10000

target:
  catalog: main
  schema: crm_curated
  table: s_customers

layer: silver
mode: hash_diff_upsert
hash_keys: [customer_id]
transform:
  cast:
    credit_limit: decimal(18,2)
    updated_at: timestamp
  standardize:
    email:
      trim: true
      lower: true
      empty_as_null: true
    status:
      trim: true
      lower: true
  derive:
    updated_date: to_date(updated_at)
  deduplicate:
    keys: [customer_id]
    order_by:
      - column: updated_at
        direction: desc
        nulls: last
schema_policy: additive_only
quality_rules:
  not_null: [customer_id]
  unique_key: [customer_id]
  expressions:
    - name: valid_status
      expression: status IN ('active', 'inactive')
      severity: quarantine

Python

from contractforge import ingest

result = ingest(
    source={
        "type": "connector",
        "connector": "postgres",
        "options": {
            "url": "{{ secret:crm/postgres_url }}",
            "dbtable": "public.customers",
            "driver": "org.postgresql.Driver",
        },
        "read": {
            "partition_column": "customer_id",
            "lower_bound": 1,
            "upper_bound": 1000000,
            "num_partitions": 8,
            "fetchsize": 10000,
        },
    },
    catalog="main",
    target_schema="crm_curated",
    target_table="s_customers",
    layer="silver",
    mode="hash_diff_upsert",
    hash_keys=["customer_id"],
    transform={
        "cast": {"credit_limit": "decimal(18,2)", "updated_at": "timestamp"},
        "standardize": {
            "email": {"trim": True, "lower": True, "empty_as_null": True},
            "status": {"trim": True, "lower": True},
        },
        "derive": {"updated_date": "to_date(updated_at)"},
        "deduplicate": {
            "keys": ["customer_id"],
            "order_by": [{"column": "updated_at", "direction": "desc", "nulls": "last"}],
        },
    },
    schema_policy="additive_only",
    quality_rules={
        "not_null": ["customer_id"],
        "unique_key": ["customer_id"],
        "expressions": [
            {"name": "valid_status", "expression": "status IN ('active', 'inactive')", "severity": "quarantine"}
        ],
    },
)

Campos de nível superior

Campo	Tipo	Padrão	Usar
_metadata	objeto		Metadados do contrato removidos antes da execução e mantidos para auditoria.
source	string | objeto	obrigatório	Nome da tabela de origem, Python DataFrame, especificação do conector ou especificação do carregador automático.
target	objeto	opcional	Objeto de destino canônico. Prefira isso a campos separados de catálogo/esquema/tabela em YAML.
target.catalog	string	main	Catálogo de destino.
target.schema	string	target_schema ou layer	Esquema de destino físico.
target.table	string	obrigatório	Nome da tabela de destino.
target_table	string	obrigatório se target.table estiver ausente	Nome da tabela de destino direto.
catalog	string	main	Catálogo de destino direto.
target_schema	string | nulo	nulo	Esquema de destino físico direto. Se ausente, layer é usado como esquema.
layer	string	bronze	Camada lógica. Não se restringe ao bronze/prata/ouro.
mode	enumeração	append	Modo de gravação.
preset / presets	string | lista[string]	desarmar	Nomes predefinidos aplicados antes da validação do plano.
applied_presets	lista[string]	[]	Nomes de presets gravados, geralmente preenchidos pelo mecanismo de preset.
source_system	string	default	Rótulo do sistema de origem gravado em colunas técnicas e tabelas de controle.
ctrl_schema	string	ops	Esquema da tabela de controle.
notebook_name	string	unknown	Rótulo de contexto de execução.
description, owner, domain, tags, sla	strings/lista	nulo/[]	Metadados leves. Prefira anotações/operações para metadados operacionais e de catálogo.
runtime_parameters	objeto		Valores de runtime de formato livre registrados com a execução.

Campos do objeto de origem

Campo	Tipo	Padrão	Usar
source.type	connector | autoloader	obrigatório	Tipo de resolvedor de origem.
source.connector	string	necessário para conector	Nome do conector nativo ou registrado.
source.name	string | nulo	nulo	Nome de origem lógica para metadados.
source.provider	string | nulo	nulo	Provedor de armazenamento de objetos: adls, azure_blob, gcs, s3.
source.format	string | nulo	nulo	Formato de resposta de arquivo, HTTP ou conector.
source.path	string | nulo	nulo	Caminho do sistema de arquivos, caminho de armazenamento de objetos, URL HTTP ou local de origem.
source.account_url	string | nulo	nulo	URL da conta do Blob do Azure.
source.container	string | nulo	nulo	Contêiner Blob/ADLS do Azure.
source.table	string | nulo	nulo	Tabela de catálogo/JDBC/conector externo.
source.query	string | nulo	nulo	Consulta SQL para conectores SQL, JDBC ou Spark externos.
source.schema	string	desarmar	Alias ​​para source.read.schema.
source.options	objeto		Opções de faísca/conector.
source.read	objeto		Controles de leitura com reconhecimento de estrutura.
source.request	objeto		Configurações de solicitação HTTP.
source.auth	objeto		Configurações de autenticação.
source.pagination	objeto		Configurações de paginação REST.
source.response	objeto		Configurações de análise de resposta REST/HTTP.
source.incremental	objeto		Configurações incrementais do lado do conector.
source.limits	objeto		Tempo limite, novas tentativas, taxa e limites de bytes.

Namespaces do conector

Espaço para nome	Campos
source.read	schema, source_complete, full_snapshot, recursiveFileLookup, pathGlobFilter, file_regex, file_regex_scope, file_regex_max_listed, file_regex_recursive, JDBC partition_column/lower_bound/upper_bound/num_partitions/fetchsize, materialização JSON staging_path/json_options, carregador automático schema_location/checkpoint_location/schema_hints/include_existing_files/max_files_per_trigger.
source.request	url, method, headers, params, json, body.
source.auth	type, token, header, value/key, username/user, password, OAuth token_url/client_id/client_secret/scope, AWS access_key_id/secret_access_key/session_token/region/credential_provider, Azure sas_token.
source.pagination	type, max_pages, page_param, start_page, offset_param, limit_param, page_size, next_cursor_path, cursor_param.
source.response	mode, raw_column, records_path, format, encoding.
source.incremental	watermark_column, initial_value, JDBC predicate, REST watermark_param/watermark_header/watermark_body_field.
source.limits	timeout_seconds, retry_attempts, retry_backoff_seconds, rate_limit_per_minute, max_pages, max_page_bytes, max_total_bytes.

Seleção, transformação e filtragem

Campo	Tipo	Padrão	Usar
select_columns	string | lista[string]	[]	Mantenha apenas as colunas de origem selecionadas.
column_mapping	objeto[string,string]		Renomeie as colunas de origem antes que as colunas técnicas da estrutura sejam adicionadas.
filter_expression	string | nulo	nulo	Filtro Spark SQL. Não combine com exclusão reversível de snapshot completo.
watermark_columns	string | lista[string]	[]	Colunas incrementais de filtro/estado.
shape	objeto | nulo	nulo	Namespace de forma direta. Prefira transform.shape.
transform.shape	objeto | nulo	nulo	Operações declarativas de análise, nivelamento, array e projeção. Consulte Transformações.
transform.cast	objeto[string,string]		Transmita colunas existentes para tipos Spark SQL após forma.
transform.standardize	objeto		Corte, normalize maiúsculas e minúsculas, normalize espaços em branco e anule strings vazias em colunas declaradas.
transform.derive	objeto[string,string]		Crie ou substitua colunas por expressões Spark SQL.
transform.deduplicate.keys	string | lista[string]	obrigatório quando existe desduplicação	Chaves de desduplicação.
transform.deduplicate.order_by	string | lista[objeto]	obrigatório quando existe desduplicação	Ordem determinística. Prefira itens estruturados com column, direction e opcional nulls.

Gravação e otimização

Campo	Tipo	Padrão	Usar
merge_keys	string | lista[string]	[]	Chaves para modos upsert, SCD2 e snapshot.
hash_keys	string | lista[string]	[]	Chaves para hash_diff_upsert.
hash_exclude_columns	string | lista[string]	[]	Colunas excluídas do hash de linha.
custom_keys	objeto		Grupos de chaves de gravador de modo/personalizado.
dedup_order_expr	string | nulo	nulo	Ordenação determinística para desduplicação de origem/destino.
partition_column	string | nulo	nulo	Coluna de partição de destino.
partition_value	string | nulo	nulo	Valor de partição única; requer partition_column.
merge_strategy	enumeração	delta	delta, delta_by_partition ou replace_partitions.
merge_partition_column	string | nulo	nulo	Coluna de partição para delta_by_partition.
replace_partitions_source_complete	booleano	falso	Evidência necessária para substituição de partição.
cluster_columns	string | lista[string]	[]	Colunas de clustering líquido/tabela quando suportadas.
zorder_columns	string | lista[string]	[]	Colunas ZORDER onde há suporte.
optimize_after_write	booleano	falso	Execute a otimização pós-gravação.
delta_properties	objeto[string,string]		Propriedades da tabela delta.

Esquema, qualidade e runtime

Campo	Tipo	Padrão	Usar
schema_policy	permissive | additive_only | strict	permissive	Manipulação do esquema de destino.
allow_type_widening	booleano	falso	Permitir alterações de ampliação suportadas.
quality_rules.required_columns	string | lista[string]	[]	Colunas de esquema obrigatórias.
quality_rules.not_null	string | lista[string]	[]	Verificações não nulas em nível de linha.
quality_rules.unique_key	string | lista[string]	[]	Verificação de exclusividade em nível de conjunto.
quality_rules.accepted_values	objeto[string,lista]		Valores permitidos por coluna.
quality_rules.min_rows	inteiro	desarmar	Contagem mínima de linhas.
quality_rules.max_null_ratio	objeto[string,número]		Proporção nula máxima por coluna.
quality_rules.expressions	lista[objeto]	[]	Regras booleanas SQL com name, expression, severity e message opcional.
quality_rules.custom	objeto		Regras de qualidade personalizadas registradas.
on_quality_fail	fail | warn | quarantine	fail	Ação herdada global para falhas de qualidade integradas.
scd2_change_columns	string | lista[string]	[]	Colunas de comparação SCD2.
scd2_effective_from_column	string | nulo	nulo	Coluna de origem efetiva do SCD2.
fix_encoding, encoding, encoding_columns	booleano/string/lista	falso/Windows-1252/[]	Reparo de codificação opcional.
dry_run	booleano	falso	Valide sem cometer efeitos colaterais de gravação/controle.
explain_mode, explain_format	booleano/enum	falso/formatted	Capture Spark explica os planos.
openlineage_enabled, openlineage_namespace, openlineage_producer	booleano/string/string	falso/nulo/contractforge	Metadados de eventos OpenLineage.
use_cache	booleano	verdadeiro	Permitir cache da estrutura.
lock_enabled	booleano	falso	Adquira bloqueio de estrutura.
idempotency_key, idempotency_policy	string/enum	nulo/always_run	Política de execução idempotente.
retry_attempts, retry_backoff_seconds	inteiro | nulo	configuração global	Tentar novamente a substituição do comportamento.
parent_run_id, run_group_id	string | nulo	nulo	IDs pai/grupo para execuções secundárias de fluxo ou janela.

Janela de execução e catchup

Campo	Tipo	Padrão	Usar
execution.window.column	string	obrigatório	Coluna filtrada por cada janela explícita.
execution.window.windows[].start, end, label	string/string/string	obrigatório/obrigatório/nulo	Limites e rótulos explícitos da janela.
execution.window.start, end, every	string	desarmar	Janelas geradas como cada 1 day.
execution.window.stop_on_failure	booleano	verdadeiro	Pare as janelas posteriores após uma falha.
execution.catchup.enabled	booleano	falso	Gere janelas de atualização a partir do estado/marca d'água salvo.
execution.catchup.column, start, end, every, stop_on_failure	misturado	nulo/nulo/nulo/nulo/verdadeiro	Configurações de geração de janela de catchup.