Monitorar o frescor de dados para tabelas de captura de dados de alterações (CDC)
Aplica-se a : 
Conectores de banco de dados
Visualização
Esse recurso está em Prévia Pública.
Este recurso está sendo lançado gradualmente. Se você não vir a coluna Data freshness em seu pipeline, o recurso não estará disponível em seu workspace.
A atualização dos dados mostra quão atuais são os dados em cada tabela de destino em comparação com o banco de dados de origem. Para cada tabela em um pipeline de conector de banco de dados, a interface do usuário do pipeline mostra quanto tempo se passou desde a alteração de origem mais recente que o pipeline ingeriu.
A frescura dos dados está disponível para fontes SQL Server, PostgreSQL, MySQL e Oracle.
Use a atualização de dados para:
- Veja o quão atrasada cada tabela de destino está em relação à origem, sem consultar o log de eventos.
- Identifique tabelas que estão ficando atrás do resto do pipeline.
- Diferencie tabelas que estão recebendo alterações ativamente de tabelas que estão ociosas na origem.
Como o frescor dos dados é calculado
A atualização dos dados é calculada para cada tabela a partir dos seguintes valores:
Valor | Definição |
|---|---|
Hora do commit de origem | O horário em que uma transação foi confirmada no banco de dados de origem. |
Hora do commit de origem mais recente | O horário do commit de origem mais recente entre as alterações que o pipeline ingeriu para a tabela. |
Atualização dos dados | O tempo atual menos o tempo do commit de origem mais recente. |
Por exemplo, se a alteração mais recente ingerida para uma tabela foi confirmada na origem há 3 minutos, a atualização dos dados da tabela é 3m. Um valor menor significa que a tabela de destino está mais próxima do estado atual da origem.
O frescor dos dados mede o quão atualizada está a tabela de destino. Diferencia-se das métricas de latência do CDC no log de eventos do gateway, como discovery_latency_ms, que medem apenas o caminho do banco de dados de origem até o volume de staging. Consulte Latência do CDC.
Atualização dos dados da view
- No workspace do Databricks, clique em Jobs & Pipelines na barra lateral.
- Clique no nome do seu pipeline de ingestão.
- Na lista de tabelas, encontre a coluna Data freshness (Atualização de dados ).
A coluna mostra um valor para cada tabela, em horas, minutos e segundos (por exemplo, 3m 23s ou 45h 32m 4s).
Como funciona a atualização de dados
- Atualizações em execução : A frescura dos dados é atualizada continuamente. O valor aumenta à medida que o tempo passa e diminui quando o pipeline ingere novas alterações da fonte.
- Atualizações interrompidas ou concluídas : a atualização de dados para de ser atualizada quando a atualização termina. A coluna continua exibindo o valor de quando a atualização terminou, para que você ainda possa ver a atualização de cada tabela após o pipeline parar.
- Tabelas paradas : a atualização dos dados reflete a última alteração que o pipeline recebeu da fonte. Se uma tabela de origem não tiver novos commit, sua atualização de dados continuará aumentando, mesmo quando o pipeline estiver íntegro e totalmente atualizado.
Interpretar a atualização dos dados
Como as tabelas de origem inativas também mostram o aumento da atualização de dados, compare o valor com a atividade na origem antes de tratá-lo como um atraso.
Padrão | O que significa |
|---|---|
Baixo e estável em todas as tabelas | O pipeline está em dia com a fonte. |
Aumentando em uma ou mais tabelas | As tabelas de origem podem estar paradas, sem novos commit. Se as tabelas de origem estiverem recebendo alterações, o pipeline estará atrasado. Consulte Encontre a origem de um atraso. |
Vazio ( | O pipeline não ingeriu nenhuma alteração para a tabela desde que começou, por exemplo, enquanto o snapshot inicial está em execução. Consulte Limitações. |
Encontre a origem de um atraso
Para pipelines que usam um gateway de ingestão, compare a última alteração de origem que o gateway leu com a última hora de commit de origem que o pipeline de ingestão ingeriu. O gateway relata event_time:max em seu log de eventos. O pipeline de ingestão relata upstream_time:max em seu log de eventos. Consulte query frescor de dados do log de eventos.
Gateway | Pipeline de ingestão | O que significa |
|---|---|---|
Avançando | Avançando | As alterações estão fluindo pelo pipeline. |
Avançando | Não avança | O pipeline de ingestão está atrás do gateway. Verifique o status do pipeline de ingestão e sua programar. |
Não avança | Não avança | A fonte está parada ou o gateway perdeu sua conexão com a fonte. Para diagnosticar o gateway, consulte Latência do CDC. |
query a atualidade dos dados do log de eventos
O log de eventos do pipeline de ingestão, não o log de eventos do gateway, registra as métricas nas quais a atualização dos dados se baseia. Faça uma query no log de eventos para rastrear a atualização dos dados ao longo do tempo, comparar tabelas ou criar alertas.
Cada evento flow_progress para um fluxo CDC relata os seguintes campos sob details:flow_progress:streaming_metrics:
campo | Tipo | Descrição |
|---|---|---|
| String | O tempo do commit de origem mais recente entre as alterações no lote, como um timestamp ISO 8601 em UTC. |
| String | O horário de commit da fonte mais antiga entre as alterações no lote, como um timestamp ISO 8601 em UTC. |
| Integer | O atraso de ponta a ponta para o lote, em milissegundos. O atraso é medido desde o tempo de commit de origem da alteração mais antiga no lote até o momento em que o lote foi feito commit na tabela de destino. |
Em cada query, substitua <ingestion-pipeline-id> pelo ID do seu pipeline de ingestão. O log de eventos do gateway não relata esses campos.
Atualidade dos dados e atraso mais recente por tabela
A query a seguir retorna o horário do commit de origem mais recente, a atualização dos dados, o horário do último lote e o atraso de ponta a ponta do lote mais recente para cada tabela. Os resultados são classificados para que as tabelas menos atualizadas apareçam primeiro.
A query inclui apenas as tabelas na atualização mais recente do pipeline. Sem este filtro, as tabelas que você removeu do pipeline permanecem no log de eventos e a frescura de seus dados continua aumentando.
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
)
SELECT
origin.flow_name AS flow_name,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp) AS latest_source_commit_time,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes,
MAX(timestamp) AS last_batch_time,
MAX_BY(details:flow_progress:streaming_metrics:stream_latency_ms::bigint, timestamp) AS latest_lag_ms
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
ORDER BY data_freshness_minutes DESC
Atraso de ponta a ponta ao longo do tempo
A seguinte query retorna o atraso de ponta a ponta máximo por hora para cada tabela nas últimas 24 horas. Use-o para descobrir quando o atraso aumentou.
SELECT
DATE_TRUNC('HOUR', timestamp) AS ts_hour,
origin.flow_name AS flow_name,
ROUND(MAX(details:flow_progress:streaming_metrics:stream_latency_ms::bigint) / 1000.0 / 60.0, 1) AS max_lag_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND details:flow_progress:streaming_metrics:stream_latency_ms IS NOT NULL
AND timestamp >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY DATE_TRUNC('HOUR', timestamp), origin.flow_name
ORDER BY origin.flow_name, ts_hour
Criar um alerta sobre o frescor de dados
Para ser notificado quando uma tabela estiver atrasada, crie um alerta Databricks SQL em uma query de frescor de dados. Por exemplo, a seguinte query retorna o número de tabelas na atualização mais recente do pipeline cuja frescor de dados é superior a 120 minutos. Substitua 120 pelo seu limite e configure o alerta para Trigger quando stale_tables for maior que 0.
WITH latest_update AS (
SELECT origin.update_id AS update_id
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1
),
current_flows AS (
SELECT DISTINCT origin.flow_name AS flow_name
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IS NOT NULL
AND origin.update_id = (SELECT update_id FROM latest_update)
),
per_table_freshness AS (
SELECT
origin.flow_name AS flow_name,
TIMESTAMPDIFF(
MINUTE,
MAX(details:flow_progress:streaming_metrics:upstream_time:max::timestamp),
current_timestamp()
) AS data_freshness_minutes
FROM event_log('<ingestion-pipeline-id>')
WHERE event_type = 'flow_progress'
AND origin.flow_name IN (SELECT flow_name FROM current_flows)
AND details:flow_progress:streaming_metrics:upstream_time:max IS NOT NULL
GROUP BY origin.flow_name
)
SELECT
COUNT_IF(data_freshness_minutes > 120) AS stale_tables
FROM per_table_freshness
As tabelas que ainda não possuem um valor de frescor de dados, como tabelas cujo snapshot inicial está sendo executado, não são contabilizadas.
A query compara o horário do commit de origem mais recente com o horário atual, então seu valor continua aumentando para tabelas de origem ociosas e enquanto o pipeline não estiver sendo executado. Ao contrário do valor na interface do usuário, ele não para de mudar quando uma atualização termina. Para evitar alertas que são disparados quando os dados estão atualizados:
- Defina o limite com base na frequência com que suas tabelas de origem recebem alterações.
- Para um pipeline executado em uma programar, defina o limite mais alto do que o intervalo entre as execuções programadas.
- Se suas tabelas recebem alterações em taxas muito diferentes, crie um alerta separado para cada grupo de tabelas com taxas semelhantes e filtre cada query em
flow_name.
Consulte Alertas do Databricks SQL e Criar um alerta.
Limitações
- Tabelas sem alterações : se uma tabela não recebeu nenhuma alteração desde que o pipeline começou, a atualização de dados estará vazia. A IU mostra um traço (
-) e o log de eventos não tem um valorupstream_timepara a tabela. Isso inclui tabelas cujo snapshot inicial está sendo executado. - Fontes SQL Server : quando o tempo de commit da fonte não está disponível para uma alteração, o gateway usa um tempo de commit em cache que ele faz refresh a cada 120 segundos. Até que o valor em cache faça refresh, o frescor de dados pode ser maior que o atraso real.
- Fontes Oracle : Para tabelas Oracle com colunas de objetos grandes (LOB), como
BLOB,CLOBouNCLOB, o frescor dos dados pode ser negativo. Um valor negativo significa que a hora de commit de origem registrada é posterior à hora atual. Isso não significa que os dados são mais recentes que a fonte. Se você vir valores negativos, entre em contato com o suporte da Databricks.