Monitoramento e query de eventos

Todo LakeFlow pipeline emite eventos que capturam logs de auditoria, verificações de qualidade de dados, progresso do pipeline e linhagem de dados. Você pode consultar esses eventos de duas fontes:

• The pipeline_events system table contains events for all pipelines across the workspaces in a region and is the recommended way to query events. Está em Beta.
• O log de eventos por pipeline é uma tabela Delta que contém eventos de um único pipeline.

You query events with standard SQL. To help you get started, this page also provides an example dashboard and common queries for the system table.

Requisitos

Para acessar esta tabela do sistema, os usuários devem:

• Ser administrador do metastore e administrador da account, ou
• Ter as permissões USE e SELECT nos esquemas do sistema. Consulte Grant access to system tables.

Painel de exemplo

Este painel de exemplo lê a tabela de sistema pipeline_events para acompanhar atualizações de pipeline, o throughput de fluxo, o backlog, a qualidade dos dados e os erros em cada pipeline de uma região. Filter every page by pipeline, table, tag, and time range.

Importar o dashboard

1. Faça download do arquivo JSON do painel.
2. Importe o painel para seu workspace. Para obter instruções, consulte Import a dashboard file.

Monitoramento queries

As seguintes queries do painel demonstram casos de uso comuns de monitoramento de pipeline.

Latest error per pipeline

This query returns the most recent error for each pipeline that errored in the last 7 days, with the outermost exception.

SQL

SELECT
  workspace_id,
  pipeline_id,
  event_time,
  message,
  error.exceptions[0].error_class AS exception_error_class,
  error.exceptions[0].sql_state   AS exception_sql_state
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
QUALIFY
  ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY event_time DESC) = 1
ORDER BY
  event_time DESC

Taxa de erros por hora

This query counts errors per pipeline per hour so you can spot spikes.

SQL

SELECT
  pipeline_id,
  date_trunc('HOUR', event_time) AS hour,
  count(*) AS num_errors
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  level = 'ERROR'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  pipeline_id,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC

Linhas alteradas por fluxo

This query sums the rows a flow appended, upserted, and deleted per hour. Each metric is a per-micro-batch count, so summing over the window gives throughput.

SQL

SELECT
  origin.flow_name,
  date_trunc('HOUR', event_time) AS hour,
  SUM(
    ifnull(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT'), 0)
    + ifnull(variant_get(details, '$.flow_progress.metrics.num_upserted_rows', 'BIGINT'), 0)
    + ifnull(variant_get(details, '$.flow_progress.metrics.num_deleted_rows', 'BIGINT'), 0)
  ) AS rows_changed
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 7 DAYS
GROUP BY
  origin.flow_name,
  date_trunc('HOUR', event_time)
ORDER BY
  hour DESC,
  rows_changed DESC

Backlog por fluxo

Essa query retorna a leitura de backlog mais recente por fluxo. Um fluxo COMPLETED está em dia, portanto, seu backlog é relatado como 0.

SQL

WITH latest_flow_progress AS (
  SELECT
    workspace_id,
    pipeline_id,
    origin.flow_name,
    event_time,
    CASE
      WHEN variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED' THEN 0
      ELSE variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')
    END AS backlog_bytes
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
    AND (
      variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT') IS NOT NULL
      OR variant_get(details, '$.flow_progress.status', 'STRING') = 'COMPLETED'
    )
  QUALIFY
    ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id, origin.flow_name ORDER BY event_time DESC) = 1
)
SELECT *
FROM latest_flow_progress
WHERE backlog_bytes > 0
ORDER BY backlog_bytes DESC

Falhas de expectativas por dataset

Esta query retorna expectativas que falharam em registros por dataset, por atualização, no último dia. Ele usa como chave a própria dataset da expectativa, que é preenchida mesmo quando a origin.dataset_name do evento não é.

SQL

SELECT
  pipeline_id,
  update_id,
  coalesce(expectation.dataset, origin.dataset_name) AS dataset_name,
  expectation.name AS expectation_name,
  SUM(expectation.failed_records) AS failed_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
  LATERAL VIEW explode(variant_get(details, '$.flow_progress.data_quality.expectations', 'ARRAY<STRUCT<name:STRING,dataset:STRING,passed_records:BIGINT,failed_records:BIGINT>>')) AS expectation
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 DAY
GROUP BY
  pipeline_id,
  update_id,
  coalesce(expectation.dataset, origin.dataset_name),
  expectation.name
HAVING
  SUM(expectation.failed_records) > 0
ORDER BY
  failed_records DESC