Referência da tabela do sistema de eventos de pipeline

Este artigo é uma referência para a pipeline_events tabela do sistema, que regista entradas de registo de eventos dos pipelines Lakeflow para pipelines na sua conta. Cada linha é um evento imutável do registo de eventos do pipeline, capturando transições do ciclo de vida, progresso do fluxo, métricas de qualidade dos dados, erros, recursos do cluster e outros dados operacionais em todos os pipelines e espaços de trabalho dentro de uma região.

Requirements

• Para aceder a esta tabela do sistema, os utilizadores devem ou:
  • Seja um administrador de metastore e um administrador de conta, ou
  • Ter permissões de USE e SELECT nos esquemas do sistema. Ver Conceder acesso às tabelas do sistema.

Tabelas de eventos do pipeline disponíveis

A tabela do sistema de eventos do pipeline permanece no lakeflow_pipeline_events_preview esquema durante a Beta e passa para o lakeflow esquema na disponibilidade geral:

Referência detalhada do esquema

Esquema da tabela de eventos do pipeline

A tabela de eventos do pipeline é apenas para acréscimos. Cada linha regista um único evento emitido por uma atualização do pipeline no momento em que foi emitida, e as linhas nunca são modificadas ou eliminadas no local.

Quais campos são preenchidos numa linha depende do tipo de evento. error, update_id, e muitos origin.* subcorpos são definidos apenas em eventos onde se aplicam, e a estrutura do details corpo também varia por event_type.

Use esta tabela para consultar a atividade histórica do oleoduto, criar alertas sobre falhas do oleoduto e correlacionar o comportamento do oleoduto com outras tabelas do sistema Lakeflow.

Caminho da tabela: system.lakeflow_pipeline_events_preview.pipeline_events

Tonalidade primária: (account_id, pipeline_event_id)

Campos de estruturas de origem

Campos de estruturas de erro

Campo de detalhes

A details coluna é um VARIANT, e os campos que contém dependem do event_type. Para os campos disponíveis em cada tipo de evento, consulte Esquema do registo de eventos do pipeline. Use a variant_get sintaxe da função ou do ponto para ler valores aninhados. Veja as consultas de exemplo abaixo para padrões típicos de acesso.

A event_type chave envolve a carga útil. Por exemplo, as métricas de um flow_progress evento estão em $.flow_progress.metrics, não $.metrics. Inclua a chave do tipo evento em cada caminho.

-- Using variant_get (lets you cast to a specific type)
SELECT
  pipeline_id,
  event_time,
  variant_get(details, '$.flow_progress.status', 'STRING')                       AS flow_status,
  variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')      AS rows_written,
  variant_get(details, '$.flow_progress.metrics.backlog_bytes', 'BIGINT')        AS backlog_bytes
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

-- Using dot syntax (returns VARIANT, cast when needed)
SELECT
  pipeline_id,
  event_time,
  details:flow_progress.status::STRING               AS flow_status,
  details:flow_progress.metrics.num_output_rows::BIGINT AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR

Exemplos de consultas

-- Flow throughput for a specific pipeline
SELECT
  origin.flow_name,
  date_trunc('HOUR', event_time) AS hour,
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) AS rows_written
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  pipeline_id = '<your-pipeline-id>'
  AND 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_written DESC

-- The latest error for each pipeline that has errored in the last 7 days, with the outermost exception.
-- The exception chain is ordered with the root cause last, so read element -1 for the root cause.
-- On many errors only the first element carries error_class and sql_state.
SELECT
  workspace_id,
  pipeline_id,
  event_time,
  event_type,
  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

-- Data quality: failed expectations by dataset, per update, in the last 1 day
SELECT
  pipeline_id,
  update_id,
  origin.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,
  origin.dataset_name,
  expectation.name
HAVING
  SUM(expectation.failed_records) > 0
ORDER BY
  failed_records DESC

Padrões comuns de junção

Junte-se à pipelines tabela para filtrar pelo nome do pipeline

A pipelines tabela é uma dimensão que muda lentamente (SCD2). Tome a versão mais recente de cada pipeline antes de aderir.

WITH latest_pipelines AS (
  SELECT *
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC) = 1
)
SELECT
  p.name AS pipeline_name,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events e
JOIN
  latest_pipelines p
  ON e.workspace_id = p.workspace_id
  AND e.pipeline_id = p.pipeline_id
WHERE
  e.level = 'ERROR'
  AND e.event_time >= current_timestamp() - INTERVAL 24 HOURS
ORDER BY
  e.event_time DESC

Juntar com pipeline_update_timeline em update_id

SELECT
  u.period_start_time AS update_start,
  u.period_end_time   AS update_end,
  e.event_time,
  e.event_type,
  e.level,
  e.message
FROM
  system.lakeflow.pipeline_update_timeline u
JOIN
  system.lakeflow_pipeline_events_preview.pipeline_events e
  ON e.update_id = u.update_id
WHERE
  u.pipeline_id = '<your-pipeline-id>'
  AND u.period_start_time >= current_timestamp() - INTERVAL 7 DAYS
ORDER BY
  u.period_start_time DESC,
  e.event_time ASC

Configuração de alertas

Podes criar alertas usando pipeline_eventsalertas SQL do Databricks. Escreve uma consulta SQL contra pipeline_events (opcionalmente ligada a outras tabelas do sistema Lakeflow), agenda-a num SQL warehouse e configura um destino de notificação (email, Slack, webhook, PagerDuty).

Alguns pontos úteis para começar:

Alerta quando não surgiram eventos para um oleoduto nos últimos N minutos

Use isto para detetar pipelines bloqueados ou a falhar silenciosamente.

-- Returns one row per pipeline that has not emitted any event in the last 30 minutes.
-- The alert can trigger when this query returns any rows.
SELECT
  workspace_id,
  pipeline_id,
  MAX(event_time) AS last_event_time
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_time >= current_timestamp() - INTERVAL 24 HOURS
GROUP BY
  workspace_id,
  pipeline_id
HAVING
  MAX(event_time) < current_timestamp() - INTERVAL 30 MINUTES

Avise quando o atraso para um fluxo específico for demasiado elevado

O backlog é reportado em flow_progress eventos como backlog_bytes, e para fontes de ficheiros também como backlog_files. Ativa-se quando a leitura mais recente ultrapassa um limiar (por exemplo, 100 MB de trabalho por processar). Nem todas as fontes reportam todas as métricas, por isso filtra pela que a tua fonte preenche.

-- Returns the most recent backlog reading per flow for a given pipeline.
-- The alert can trigger when backlog_bytes exceeds the threshold for any flow.
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
    pipeline_id = '<your-pipeline-id>'
    AND 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 > 100000000

Para ver qual fonte está por trás, $.flow_progress.metrics.source_metrics é um conjunto de leituras por fonte, cada uma com source_name o de backlog_bytes, backlog_records ou backlog_files.

Alerta sobre quedas na qualidade dos dados num pipeline

Cada flow_progress evento reporta o número de carreiras descartadas segundo EXPECT … DROP as expectativas. Soma-os por conjunto de dados numa janela de atualização e alerta quando o total ultrapassar um limiar.

-- Returns datasets where more than 100 rows were dropped by expectations, per update, in the last hour.
SELECT
  pipeline_id,
  update_id,
  origin.dataset_name,
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) AS dropped_records
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR
GROUP BY
  pipeline_id,
  update_id,
  origin.dataset_name
HAVING
  SUM(variant_get(details, '$.flow_progress.data_quality.dropped_records', 'BIGINT')) > 100

Alerta quando um fluxo processa poucas linhas

flow_progress os eventos reportam metrics.num_output_rows como uma contagem por micro-lote, por isso, somando os eventos numa janela, as linhas escritas sobre essa janela. Crie um alerta para quando o débito descer abaixo de um mínimo esperado. Por exemplo, um fluxo que normalmente escreve milhares de linhas por hora mas produz quase zero pode indicar uma fonte mal configurada.

Esta consulta apenas reporta fluxos que emitiram um flow_progress evento com contagem de linhas na janela. Um fluxo totalmente parado não emite eventos, por isso emparelhe este alerta com o alerta de eventos em falta acima.

-- Returns flows that wrote fewer than 100 rows in the last hour.
-- The alert can trigger when this query returns any rows.
SELECT
  pipeline_id,
  origin.flow_name,
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) AS rows_written_last_hour
FROM
  system.lakeflow_pipeline_events_preview.pipeline_events
WHERE
  pipeline_id = '<your-pipeline-id>'
  AND event_type = 'flow_progress'
  AND event_time >= current_timestamp() - INTERVAL 1 HOUR
  AND variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT') IS NOT NULL
GROUP BY
  pipeline_id,
  origin.flow_name
HAVING
  SUM(variant_get(details, '$.flow_progress.metrics.num_output_rows', 'BIGINT')) < 100

Alerte quando a latência de novos dados for demasiado elevada

Para fluxos de streaming, flow_progress os eventos reportam latência em streaming_metrics. stream_latency_ms é o tempo desde que os dados chegaram a montante até ao momento em que o micro-batch se comprometeu na tabela Delta. Pode definir um gatilho para quando a leitura mais recente ultrapassa um limiar (por exemplo, 5 minutos).

Apenas o streaming flui com um relatório stream_latency_msde tempo de evento marcado, e apenas quando as métricas de tempo SDP estão ativadas. Outros fluxos regressam NULL a cada evento, e este alerta nunca é ativado para eles.

-- Returns the most recent new-data latency per flow for a given pipeline.
-- The alert can trigger when stream_latency_ms > 300000 (5 minutes) for any flow.
WITH latest_latency 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.streaming_metrics.stream_latency_ms', 'BIGINT')
    END AS stream_latency_ms
  FROM
    system.lakeflow_pipeline_events_preview.pipeline_events
  WHERE
    pipeline_id = '<your-pipeline-id>'
    AND event_type = 'flow_progress'
    AND event_time >= current_timestamp() - INTERVAL 1 HOUR
    AND (
      variant_get(details, '$.flow_progress.streaming_metrics.stream_latency_ms', '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_latency
WHERE stream_latency_ms > 300000

Dicas para alertas de produção

• Filtra por pipeline_id (e workspace_id se mantiveres alertas por espaço de trabalho) para que cada alerta tenha como alvo um âmbito específico em vez de toda a conta.
• Escolha uma cadência de avaliação que corresponda à sensibilidade do alerta. Use um intervalo curto (por exemplo, a cada 5 minutos) para sinais de falha rápida e um intervalo mais longo (por exemplo, horário) para atrasos e tendências de qualidade dos dados. Uma condição de "disparo quando a consulta retorna mais de 0 linhas" funciona na maioria dos casos.

Valores de referência

Valores dos níveis

Valores do nível de maturidade

Valores do tipo de evento

O event_type campo é uma enumeração. O conjunto completo de valores: