Operar a ingestão por meio de evidências de tabela de controle.

ContractForge grava evidências operacionais em tabelas de controle Delta para que as equipes possam diagnosticar falhas, rastrear atualizações, auditar governança, inspecionar lotes de streaming, estimar o custo do runtime e verificar versões da estrutura sem depender da saída do notebook.

Modelo de esquema de controle

As tabelas de controle são criadas por ensure_ctrl_tables() no catálogo de controle e esquema configurados. Eles são aditivos: as atualizações da estrutura adicionam colunas conhecidas quando ausentes e não removem colunas existentes automaticamente.

Nomes padrão

Nomes de tabelas estáveis

O prefixo padrão é ctrl_ingestion_*. Os nomes podem ser alterados por meio de FrameworkConfig, mas os painéis devem preferencialmente resolver nomes do esquema de controle configurado.

Versionamento

Estrutura e esquema de controle

A maioria das tabelas operacionais inclui framework_version e ctrl_schema_version. Use-os para detectar versões mistas e layouts de metadados desatualizados.

Partições

Tabelas de alto volume particionadas por data

ctrl_ingestion_runs e ctrl_ingestion_errors são particionados por data. Use filtros de data em painéis e trabalhos de retenção.

Cargas JSON

Detalhes operacionais flexíveis

Opções de conector, metadados de origem, durações de estágio, alterações de esquema e métricas de operação são armazenados como strings JSON para manter o esquema da tabela estável.

Mapa de tabela

Mesa	Propósito	Questões operacionais primárias
ctrl_ingestion_runs	Uma linha de execução em lote por execução, incluindo origem, destino, status, métricas, runtime, esquema e resumos de governança.	O que correu? Conseguiu? Quantas linhas foram movidas? Qual conector, modo, runtime e versão foram usados?
ctrl_ingestion_state	Estado mais recente por tabela de destino, incluindo marca d'água, último sucesso, último erro e última versão Delta.	Qual é a marca d'água atual? Quando foi a última vez que a mesa teve sucesso? O estado avançou?
ctrl_ingestion_quality	Resultados de regras de qualidade para regras com falha, avisos, quarentena ou anulação.	Quais regras falharam? Quantos registros falharam? A corrida foi abortada ou colocada em quarentena?
ctrl_ingestion_quarantine	Cargas úteis de linha em quarentena com regra e motivo.	Quais registros foram isolados e por quê?
ctrl_ingestion_errors	Detalhes completos de falhas e rastreamentos de pilha.	Qual é a verdadeira exceção por trás de um erro de curto prazo?
ctrl_ingestion_schema_changes	Eventos de evolução de esquema detectados e aplicados.	Quais colunas foram adicionadas? Quais alterações de tipo foram detectadas ou rejeitadas?
ctrl_ingestion_streams	Linha de execução de streaming externo ou carregador automático com métricas agregadas de microlote.	Quantos lotes foram executados? O agregado correspondeu aos totais das corridas secundárias?
ctrl_ingestion_locks	Estado de bloqueio de melhor esforço por destino.	Um alvo está atualmente bloqueado? Quem é o dono da fechadura? Ele expirou?
ctrl_ingestion_explain	Plano de explicação do Spark capturado quando ativado.	Qual plano do Spark foi usado em uma simulação, verificação de CI ou execução de diagnóstico?
ctrl_ingestion_lineage	Cargas de eventos compatíveis com OpenLineage.	Qual fonte produziu qual alvo e qual evento de linhagem foi emitido?
ctrl_ingestion_metadata	Metadados da estrutura do plano de controle.	Qual versão da estrutura e versão do esquema de controle estão instaladas?
ctrl_ingestion_annotations	Comentários de catálogo, tags, aliases, PII e ações de anotação de suspensão de uso.	Quais metadados foram aplicados, ignorados, avisados ​​ou falharam?
ctrl_ingestion_operations	Propriedade declarada, SLA, criticidade, alertas e instantâneos de metadados de suporte.	Quem é o dono da mesa? Qual é o frescor esperado? Qual runbook se aplica?
ctrl_ingestion_access	Acesse evidências de validação/aplicação de contrato para concessões, filtros de linha e máscaras.	Quais alterações de acesso foram aplicadas ou adiadas? A deriva foi detectada?

Referência de campo

Use esta seção como dicionário de dados operacionais para painéis, regras de alerta e consultas de suporte. Os campos JSON são armazenados como strings; analise-os com funções from_json, get_json_object ou SQL JSON ao criar visualizações operacionais selecionadas.

ctrl_ingestion_runs — razão de execução em lote

Tabela de fatos central para ingestão em lote. Junte-se a erros, qualidade, alterações de esquema, anotações e acesso por run_id.

Campo	Significado
run_id	Identificador de execução exclusivo para esta execução em lote.
run_ts_utc	Carimbo de data/hora de execução usado como a hora do evento lógico para a linha.
run_date	Partição de data derivada do carimbo de data/hora de execução.
notebook_name	Caderno ou nome do chamador, quando disponível.
layer	Camada lógica declarada no contrato, como bronze, prata, ouro, estágio ou valor personalizado.
source_table	Identificador de tabela de origem herdado ou resolvido quando a origem é semelhante a uma tabela.
source_type	Categoria de origem, por exemplo tabela, sql, caminho, conector ou fluxo.
source_connector	Nome do conector, como jdbc, s3, azure_blob, rest_api, http_file ou autoloader.
source_name	Nome de origem lógica do contrato de origem.
source_provider	Provedor ou plataforma da fonte, quando declarado.
source_format	Formato de entrada como csv, json, parquet, avro, delta, xml, text ou jdbc.
source_path	Caminho ou URL usado pelos conectores de armazenamento de arquivo/objeto.
source_options_json	Opções de conector redigidas.
source_read_json	Configuração de leitura, como esquema, filtros glob, recursão, particionamento ou opções de leitura específicas do conector.
source_request_json	Solicitar configuração para fontes HTTP/REST.
source_auth_json	Configuração de autenticação redigida.
source_pagination_json	Configuração de paginação para fontes estilo REST.
source_response_json	Configuração de extração de resposta para fontes estilo REST.
source_incremental_json	Configurações de fonte incrementais, quando presentes.
source_limits_json	Limites de segurança do conector, como tempo limite, contagem de páginas ou limites de tamanho de carga útil.
source_capabilities_json	Capacidades declaradas ou inferidas do conector.
source_metrics_json	Métricas de leitura do lado do conector, quando disponíveis.
target_table	Tabela de destino totalmente qualificada.
mode	Modo de gravação, como append, overwrite, upsert, hash_diff_upsert, historical ou snapshot_reconcile_soft_delete.
status	Status final da execução: SUCCESS, FAILED, ABORTED ou SKIPPED.
rows_read	Número lógico de linhas de origem processadas pela estrutura.
rows_written	Número lógico de linhas gravadas no destino.
rows_inserted	Contagem lógica de linhas inseridas quando o modo pode relatá-la.
rows_updated	Contagem lógica de linhas atualizada quando o modo pode relatá-la.
rows_deleted	Contagem de linhas lógicas excluídas ou excluídas de forma reversível, quando aplicável.
rows_quarantined	Linhas gravadas na quarentena em vez do destino.
watermark_column	Coluna de marca d'água usada para leituras incrementais.
watermark_previous	Valor da marca d'água confirmada anteriormente.
watermark_current	Novo valor de marca d'água confirmado para execuções bem-sucedidas.
started_at_utc	Execute o carimbo de data e hora de início.
finished_at_utc	Execute o carimbo de data/hora de término.
duration_seconds	Tempo total de execução decorrido em segundos.
quality_status	Resumo de qualidade: NOT_CONFIGURED, PASSED, WARNED, QUARANTINED ou FAILED.
schema_policy	Política de esquema usada pela execução.
schema_changes_json	Resumo estruturado de alteração de esquema retornado pela validação/evolução de esquema.
stage_durations_json	Temporizações por estágio, como leitura, qualidade, gravação, estado e linhagem.
contract_description	Instantâneo de descrição em nível de contrato.
contract_owner	Instantâneo do proprietário no nível do contrato.
contract_domain	Instantâneo de negócios ou domínio de dados.
contract_tags_json	Tags de contrato serializadas como JSON.
contract_sla	Valor do SLA do contrato, se declarado no contrato de ingestão.
runtime_parameters_json	Instantâneo dos parâmetros de runtime.
operation_metrics_json	Métricas de gravação delta e lógica serializadas como JSON.
write_started_at_utc	Escreva o carimbo de data/hora de início do estágio.
write_finished_at_utc	Escreva o carimbo de data/hora de término do estágio.
delta_version_before	Versão da tabela delta antes da gravação, quando conhecida.
delta_version_after	Versão da tabela delta após a gravação, quando conhecida.
write_committed	Se a gravação atingiu o caminho de confirmação de destino.
error_message	Mensagem de diagnóstico curta. Use ctrl_ingestion_errors para rastreamentos completos de pilha.
parent_run_id	Identificador de execução de fluxo pai para execuções secundárias de microlotes.
run_group_id	Identificador de grupo opcional para execuções relacionadas.
master_job_id	ID do trabalho do Databricks quando disponível.
master_run_id	Os databricks executam o ID quando disponível.
idempotency_key	Chave de idempotência fornecida pelo usuário.
idempotency_policy	Política usada para tratamento de execução duplicada.
skip_reason	Motivo das execuções SKIPPED.
skipped_by_run_id	Corrida anterior que causou o salto.
metrics_source	Origem da métrica: lógica, delta ou mista.
framework_version	Versão do pacote ContractForge usada pela execução.
ctrl_schema_version	Versão do esquema de controle usada pela execução.
runtime_type	Classificação de runtime, como clássico ou serverless.
spark_version	Versão de runtime do Spark.
python_version	Versão de runtime do Python.
annotations_status	Resumo do aplicativo de anotação para a execução.
annotations_result_json	Carga útil do resultado da anotação detalhada.
ownership_json	Instantâneo de metadados de propriedade do contrato de operações.
operations_json	Instantâneo do contrato de operações.

ctrl_ingestion_state — estado de destino mais recente

Estado operacional atual por tabela de destino. As atualizações de marca d'água e versão Delta acontecem somente quando a execução é confirmada com sucesso.

Campo	Significado
target_table	Tabela de destino totalmente qualificada e chave lógica para a linha de estado.
watermark_column	Coluna de marca d'água atualmente rastreada para o destino.
watermark_value	Último valor de marca d'água confirmado.
last_success_at_utc	Carimbo de data/hora da última execução bem-sucedida.
last_run_id	ID de execução mais recente tocando neste alvo.
last_status	Status da execução mais recente que tocou esse alvo.
last_rows_written	Linhas escritas pela execução mais recente.
last_error_message	Erro curto da execução com falha mais recente.
parent_run_id	O fluxo pai é executado quando a atualização de estado mais recente vem de um microlote.
run_group_id	ID do grupo para execuções relacionadas.
master_job_id	ID do trabalho do Databricks quando disponível.
master_run_id	Os databricks executam o ID quando disponível.
last_delta_version	Última versão Delta de destino conhecida após gravação bem-sucedida.
last_write_completed_at_utc	Carimbo de data/hora quando a última gravação foi concluída.
last_watermark_candidate	Marca d'água do candidato observada durante a última execução.
last_updated_at_utc	Carimbo de data e hora em que esta linha de estado foi atualizada.

Tabelas de qualidade, quarentena e esquema

Mesa	Campo	Significado
ctrl_ingestion_quality	run_id	Execute que produziu o resultado de qualidade.
ctrl_ingestion_quality	target_table	Tabela de destino avaliada pela regra.
ctrl_ingestion_quality	rule_name	Identificador de regra, nome de regra integrado ou nome de expressão.
ctrl_ingestion_quality	status	Resultado da regra.
ctrl_ingestion_quality	severity	Ação configurada: avisar, colocar em quarentena ou abortar.
ctrl_ingestion_quality	failed_count	Número de linhas ou verificações que falharam.
ctrl_ingestion_quality	checked_at_utc	Carimbo de data e hora da avaliação da qualidade.
ctrl_ingestion_quality	message	Mensagem de regra, gerada ou personalizada.
ctrl_ingestion_quality	details_json	Detalhes adicionais da regra.
ctrl_ingestion_quarantine	run_id	Execute que colocou a linha em quarentena.
ctrl_ingestion_quarantine	target_table	Tabela de destino à qual a linha se destinava.
ctrl_ingestion_quarantine	rule_name	Regra que causou quarentena.
ctrl_ingestion_quarantine	error_reason	Motivo de quarentena legível por humanos.
ctrl_ingestion_quarantine	record_payload	Registro em quarentena serializado como JSON.
ctrl_ingestion_quarantine	quarantined_at_utc	Carimbo de data e hora da quarentena.
ctrl_ingestion_schema_changes	run_id	Execute que detectou a alteração do esquema.
ctrl_ingestion_schema_changes	change_ts_utc	Altere o carimbo de data/hora de detecção.
ctrl_ingestion_schema_changes	target_table	Tabela de destino afetada pela alteração.
ctrl_ingestion_schema_changes	change_type	Altere a classe, como add_column ou type_change.
ctrl_ingestion_schema_changes	column_name	Coluna afetada pela alteração.
ctrl_ingestion_schema_changes	source_type	Tipo de dados de origem observado.
ctrl_ingestion_schema_changes	target_type	Tipo de dados de destino observado.
ctrl_ingestion_schema_changes	applied	Se a alteração foi aplicada à tabela Delta.
ctrl_ingestion_schema_changes	details_json	Carga útil de alteração completa.
ctrl_ingestion_schema_changes	framework_version	Versão da estrutura que registrou a alteração.
ctrl_ingestion_schema_changes	ctrl_schema_version	Versão do esquema de controle.

Diagnóstico, linhagem e tabelas de streaming

Mesa	Campos	Significado
ctrl_ingestion_errors	run_id, error_ts_utc, error_date	Identidade da falha e partição de data.
ctrl_ingestion_errors	target_table, source_table, mode, status	Contexto para a execução com falha.
ctrl_ingestion_errors	error_type, error_message, stack_trace	Tipo de exceção, mensagem curta e rastreamento completo.
ctrl_ingestion_errors	framework_version, ctrl_schema_version, runtime_type, spark_version, python_version	Contexto de runtime e versão.
ctrl_ingestion_streams	stream_run_id, parent_run_id, run_group_id	Identidade e agrupamento do fluxo externo.
ctrl_ingestion_streams	idempotency_key, idempotency_policy, skip_reason, skipped_by_stream_run_id	Comportamento de idempotência para execuções de streaming.
ctrl_ingestion_streams	target_table, target_catalog, target_layer, notebook_name	Metadados de destino e chamador.
ctrl_ingestion_streams	source_type, source_path, trigger, checkpoint_location	Fonte de streaming e configuração de checkpoint.
ctrl_ingestion_streams	status, started_at_utc, ended_at_utc, duration_seconds	Tempo do ciclo de vida e status final.
ctrl_ingestion_streams	batches_processed, total_rows_read, total_rows_written, total_rows_quarantined	Métricas agregadas de microlote filho.
ctrl_ingestion_streams	framework_version, ctrl_schema_version, runtime_type, spark_version, python_version, error_message, master_job_id, master_run_id	Runtime, versão, falha e contexto do trabalho.
ctrl_ingestion_explain	run_id, target_table, source_table, mode, explain_format, plan_text, captured_at_utc	Plano Spark capturado para execuções de diagnóstico.
ctrl_ingestion_lineage	run_id, event_time_utc, event_type, target_table, source_table, namespace, producer, event_json	Evidência de evento compatível com OpenLineage.
ctrl_ingestion_metadata	component, framework_version, ctrl_schema_version, updated_at_utc	Linha de metadados do plano de controle para o estado da estrutura instalada.

Tabelas de governança e acesso

Mesa	Campos	Significado
ctrl_ingestion_annotations	run_id, target_table	Execute e direcione para a ação de metadados.
ctrl_ingestion_annotations	annotation_scope, annotation_type, column_name, key	Escopo da anotação e campo lógico, como tag de tabela, comentário de coluna, PII ou suspensão de uso.
ctrl_ingestion_annotations	previous_value, value, status, error_message, applied_sql	Evidência antes/depois, resultado, diagnóstico e ação SQL.
ctrl_ingestion_annotations	annotation_ts_utc, annotation_date, framework_version, ctrl_schema_version	Contexto de tempo e versão.
ctrl_ingestion_operations	run_id, target_table	Execute e direcione o instantâneo de metadados de operações.
ctrl_ingestion_operations	criticality, expected_frequency, freshness_sla_minutes	SLA e metadados de gravidade usados ​​por painéis e roteamento de alertas.
ctrl_ingestion_operations	alert_on_failure, alert_on_quality_fail, runbook_url	Alertas externos e metadados de resposta. ContractForge registra isso; os sistemas de alerta o consomem.
ctrl_ingestion_operations	ownership_json, owners_json, groups_json, tags_json	Propriedade, grupos de suporte e tags personalizadas serializadas como JSON.
ctrl_ingestion_operations	status, recorded_at_utc, framework_version, ctrl_schema_version	Resultado da gravação, carimbo de data/hora e contexto da versão.
ctrl_ingestion_access	access_run_id, run_id, target_table	Acesse a identidade e o destino do fluxo de trabalho.
ctrl_ingestion_access	access_type, principal, privilege	Concessão, filtro de linha, máscara de coluna ou ação de validação e principal/privilégio afetado.
ctrl_ingestion_access	column_name, function_name, object_name	Coluna, função ou objeto afetado pela governança de acesso.
ctrl_ingestion_access	status, error_message, applied_sql	Resultado, diagnóstico e comando SQL.
ctrl_ingestion_access	previous_value, new_value, mode, drift_policy, revoke_unmanaged	Contexto de deriva e reconciliação.
ctrl_ingestion_access	access_ts_utc, access_date, framework_version, ctrl_schema_version	Contexto de tempo e versão.

ctrl_ingestion_locks — coordenação de melhor esforço

Os bloqueios reduzem colisões previsíveis entre redatores cooperativos do ContractForge. Eles não substituem a simultaneidade otimista do Delta.

Campo	Significado
target_table	Tabela de destino protegida pelo bloqueio.
run_id	Execute o que possui atualmente ou já possui o bloqueio.
owner	Proprietário lógico ou identidade do chamador.
acquired_at_utc	Carimbo de data e hora de aquisição de bloqueio.
expires_at_utc	Carimbo de data e hora de expiração TTL.
ttl_minutes	Bloqueio TTL configurado em minutos.
released_at_utc	Liberação do carimbo de data/hora, quando liberado de forma limpa.
status	Status de bloqueio como ACTIVE, RELEASED ou EXPIRED.

Consultas SQL úteis

Substitua main.ops pelo catálogo e esquema usados ​​para tabelas de controle.

Integridade da execução recente

SELECT
  status,
  target_table,
  mode,
  source_connector,
  runtime_type,
  rows_read,
  rows_written,
  rows_quarantined,
  duration_seconds,
  error_message,
  run_id
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY started_at_utc DESC
LIMIT 100;

Triagem de falhas com full stack trace

SELECT
  r.started_at_utc,
  r.target_table,
  r.source_connector,
  r.mode,
  r.runtime_type,
  r.error_message AS short_error,
  e.error_type,
  e.stack_trace
FROM main.ops.ctrl_ingestion_runs r
LEFT JOIN main.ops.ctrl_ingestion_errors e
  ON r.run_id = e.run_id
WHERE r.status = 'FAILED'
  AND r.run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY r.started_at_utc DESC;

Frescura e desvio de marca d'água

SELECT
  target_table,
  watermark_column,
  watermark_value,
  last_success_at_utc,
  last_status,
  last_rows_written,
  last_delta_version,
  current_timestamp() - last_success_at_utc AS time_since_success
FROM main.ops.ctrl_ingestion_state
ORDER BY last_success_at_utc ASC NULLS FIRST;

Regras de qualidade que requerem atenção

SELECT
  q.target_table,
  q.rule_name,
  q.severity,
  q.status,
  q.failed_count,
  q.message,
  r.run_id,
  r.started_at_utc
FROM main.ops.ctrl_ingestion_quality q
JOIN main.ops.ctrl_ingestion_runs r
  ON q.run_id = r.run_id
WHERE r.run_date >= current_date() - INTERVAL 14 DAYS
  AND q.failed_count > 0
ORDER BY r.started_at_utc DESC, q.failed_count DESC;

Volume de quarentena por destino e regra

SELECT
  target_table,
  rule_name,
  count(*) AS quarantined_records,
  max(quarantined_at_utc) AS latest_quarantine
FROM main.ops.ctrl_ingestion_quarantine
GROUP BY target_table, rule_name
ORDER BY quarantined_records DESC;

Mudanças de esquema por tabela

SELECT
  target_table,
  change_type,
  column_name,
  source_type,
  target_type,
  applied,
  change_ts_utc,
  details_json
FROM main.ops.ctrl_ingestion_schema_changes
ORDER BY change_ts_utc DESC;

Streaming agregado versus execuções secundárias

WITH child AS (
  SELECT
    parent_run_id AS stream_run_id,
    count(*) AS child_batches,
    sum(coalesce(rows_read, 0)) AS child_rows_read,
    sum(coalesce(rows_written, 0)) AS child_rows_written,
    sum(coalesce(rows_quarantined, 0)) AS child_rows_quarantined
  FROM main.ops.ctrl_ingestion_runs
  WHERE parent_run_id IS NOT NULL
  GROUP BY parent_run_id
)
SELECT
  s.stream_run_id,
  s.status,
  s.batches_processed,
  c.child_batches,
  s.total_rows_read,
  c.child_rows_read,
  s.total_rows_written,
  c.child_rows_written,
  s.total_rows_quarantined,
  c.child_rows_quarantined
FROM main.ops.ctrl_ingestion_streams s
LEFT JOIN child c
  ON s.stream_run_id = c.stream_run_id
ORDER BY s.started_at_utc DESC;

Estágios lentos a partir da duração dos estágios

SELECT
  run_id,
  target_table,
  mode,
  duration_seconds,
  get_json_object(stage_durations_json, '$.read') AS read_seconds,
  get_json_object(stage_durations_json, '$.quality') AS quality_seconds,
  get_json_object(stage_durations_json, '$.write') AS write_seconds,
  get_json_object(stage_durations_json, '$.state') AS state_seconds,
  get_json_object(stage_durations_json, '$.lineage') AS lineage_seconds
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 7 DAYS
ORDER BY duration_seconds DESC
LIMIT 50;

Estimativa de rendimento e custo operacional

WITH rates AS (
  SELECT 2.0 AS dbu_per_hour, 0.40 AS currency_per_dbu
)
SELECT
  r.target_table,
  r.mode,
  r.source_connector,
  r.rows_read,
  r.rows_written,
  r.duration_seconds,
  CASE WHEN r.duration_seconds > 0 THEN r.rows_written / r.duration_seconds END AS rows_written_per_second,
  (r.duration_seconds / 3600.0) * rates.dbu_per_hour AS estimated_dbu,
  (r.duration_seconds / 3600.0) * rates.dbu_per_hour * rates.currency_per_dbu AS estimated_cost
FROM main.ops.ctrl_ingestion_runs r
CROSS JOIN rates
WHERE r.status = 'SUCCESS'
  AND r.run_date >= current_date() - INTERVAL 30 DAYS
ORDER BY estimated_cost DESC;

Status do aplicativo de governança

SELECT
  target_table,
  annotation_scope,
  annotation_type,
  column_name,
  key,
  status,
  error_message,
  annotation_ts_utc
FROM main.ops.ctrl_ingestion_annotations
ORDER BY annotation_ts_utc DESC;

Desvio de acesso e operações de segurança com falha

SELECT
  target_table,
  access_type,
  principal,
  privilege,
  column_name,
  function_name,
  mode,
  drift_policy,
  status,
  error_message,
  access_ts_utc
FROM main.ops.ctrl_ingestion_access
WHERE status <> 'APPLIED'
ORDER BY access_ts_utc DESC;

Bloqueios ativos ou expirados

SELECT
  target_table,
  run_id,
  owner,
  status,
  acquired_at_utc,
  expires_at_utc,
  released_at_utc,
  CASE
    WHEN status = 'ACTIVE' AND expires_at_utc < current_timestamp() THEN 'EXPIRED_BUT_NOT_RELEASED'
    ELSE status
  END AS effective_status
FROM main.ops.ctrl_ingestion_locks
ORDER BY acquired_at_utc DESC;

Inventário de versão da estrutura

SELECT
  framework_version,
  ctrl_schema_version,
  runtime_type,
  spark_version,
  python_version,
  count(*) AS runs
FROM main.ops.ctrl_ingestion_runs
WHERE run_date >= current_date() - INTERVAL 30 DAYS
GROUP BY framework_version, ctrl_schema_version, runtime_type, spark_version, python_version
ORDER BY runs DESC;

Fluxos de trabalho operacionais

Painel diário de saúde

Baseie-o em ctrl_ingestion_runs, agrupado por status, target_table, mode, source_connector e runtime_type. Inclui mensagens de erro recentes e links para executar IDs.

Monitoramento de frescor

Use ctrl_ingestion_state.last_success_at_utc mais ctrl_ingestion_operations.freshness_sla_minutes. Alerta fora do ContractForge; a biblioteca registra os metadados e evidências.

Resposta de qualidade

Use ctrl_ingestion_quality para resumo em nível de regra e ctrl_ingestion_quarantine para correção em nível de registro. As regras de anulação e as regras de quarentena exigem caminhos de resposta diferentes.

Suporte de runtime

Comece com ctrl_ingestion_runs.error_message, junte ctrl_ingestion_errors para rastreamento completo e verifique framework_version, runtime_type e metadados do conector antes de alterar o código.

Comandos de manutenção

Os relatórios de retenção e custos são utilitários operacionais. Os relatórios de custos estimam o custo do runtime a partir de taxas fornecidas pelo usuário; eles não substituem as exportações de faturamento na nuvem.

contractforge maintenance ctrl-retention --catalog main --ctrl-schema ops --retention-days 30
contractforge maintenance cost-report --catalog main --ctrl-schema ops --dbu-per-hour 2 --currency-per-dbu 0.40

Orientação de retenção

Aplique a retenção primeiro a tabelas particionadas por data de alto volume, como execuções e erros. Mantenha o histórico de quarentena e alterações de esquema de acordo com a política de auditoria e correção de qualidade de dados da organização.