Monitorar pipelines do Delta Live Tables

Artigo
05/03/2024

Este artigo descreve como você poderá usar recursos integrados de monitoramento e observabilidade para pipelines das Tabelas Dinâmicas Delta, incluindo linhagem de dados, histórico de atualizações e relatórios de qualidade de dados.

Você pode examinar a maioria dos dados de monitoramento manualmente por meio da interface do usuário de detalhes do pipeline. Algumas tarefas são mais fáceis de realizar consultando os metadados do log de eventos. Confira O que é o log de eventos do Delta Live Tables?.

Quais detalhes do pipeline estão disponíveis na interface do usuário?

O gráfico de pipeline é exibido assim que uma atualização de um pipeline é iniciada com êxito. As setas representam dependências entre os conjuntos de dados no pipeline. Por padrão, a página de detalhes do pipeline mostra a atualização mais recente da tabela, mas você pode selecionar atualizações mais antigas em um menu suspenso.

Os detalhes exibidos incluem a ID do pipeline, bibliotecas de origem, custo de computação, edição do produto e o canal configurado para o pipeline.

Para ver uma exibição tabular de conjuntos de dados, clique na guia Lista. A exibição Lista permite que você veja todos os conjuntos de dados em seu pipeline representados como uma linha em uma tabela e é útil quando o DAG do pipeline é muito grande para visualizar no modo de exibição do Graph. Você pode controlar os conjuntos de dados exibidos na tabela usando vários filtros, como nome, tipo e status do conjunto de dados. Para alternar de volta para a visualização do DAG, clique em Graph.

O usuário em Executar como é o proprietário do pipeline, e as atualizações de pipeline são executadas com as permissões desse usuário. Para alterar o usuário run as, clique em Permissões e altere o proprietário do pipeline.

Como você pode exibir os detalhes do conjunto de dados?

Clicar em um conjunto de dados no gráfico de pipeline ou na lista de conjuntos de dados exibe detalhes sobre o conjunto de dados. Os detalhes incluem o esquema do conjunto de dados, as métricas de qualidade dos dados e um link de volta para o código-fonte que define o conjunto de dados.

Exibir histórico de atualização

Para exibir o histórico e o status das atualizações de pipeline, clique no menu suspenso de histórico de atualizações na barra superior.

Para exibir o grafo, os detalhes e os eventos de uma atualização, selecione a atualização no menu suspenso. Para retornar à atualização mais recente, clique em Mostrar a atualização mais recente.

Obter notificações de eventos de pipeline

Para receber notificações em tempo real sobre eventos do pipeline, como a conclusão bem-sucedida de uma atualização do pipeline ou a falha de uma atualização do pipeline, adicione Adicionar notificações por email para eventos do pipeline ao criar ou editar um pipeline.

O que é o log de eventos do Delta Live Tables?

O log de eventos do Delta Live Tables contém todas as informações relacionadas a um pipeline, incluindo logs de auditoria, verificações de qualidade de dados, progresso do pipeline e linhagem de dados. Use o log de eventos para acompanhar, entender e monitorar o estado dos seus pipelines de dados.

Você pode exibir entradas de log de eventos na interface do usuário do Delta Live Tables, na API do Delta Live Tables ou consultando diretamente o log de eventos. Esta seção se concentra em consultar o log de eventos diretamente.

Você também pode definir ações personalizadas a serem executadas quando os eventos são registrados, por exemplo, enviando alertas, com ganchos de evento.

Esquema de log de eventos

A tabela a seguir descreve o esquema do log de eventos. Alguns desses campos contêm documentos JSON que exigem a análise para execução de algumas consultas, como o campo details. O Azure Databricks dá suporte ao operador : para analisar campos JSON. Confira : operador (colon sign).

Campo	Descrição
`id`	Um identificador exclusivo para o registro do log de eventos.
`sequence`	Um documento JSON que contém metadados para identificar e ordenar eventos.
`origin`	Um documento JSON que contém metadados para a origem do evento, por exemplo, o provedor de nuvem, a região do provedor de nuvem, `user_id`, `pipeline_id` ou `pipeline_type` para mostrar onde o pipeline foi criado, seja `DBSQL` ou `WORKSPACE`.
`timestamp`	A hora em que o evento foi registrado.
`message`	Uma mensagem legível por humanos que descreve o evento.
`level`	O tipo de evento, por exemplo, `INFO`, `WARN`, `ERROR` ou `METRICS`.
`error`	Em caso de erro, detalhes que descrevem o erro.
`details`	Um documento JSON que contém detalhes estruturados do evento. Esse é o campo primário usado para analisar eventos.
`event_type`	O tipo de evento.
`maturity_level`	A estabilidade do esquema de evento. Os valores possíveis são: * `STABLE`: o esquema é estável e não será alterado. * `NULL`: o esquema é estável e não será alterado. O valor poderá ser `NULL` se o registro tiver sido criado antes da adição do campo `maturity_level` (versão 2022.37). * `EVOLVING`: o esquema não é estável e pode ser alterado. * `DEPRECATED`: o esquema foi preterido e o runtime do Delta Live Tables pode parar de produzir esse evento a qualquer momento.

Consultando o log de eventos

O local do log de eventos e a interface para consultar o log de eventos dependem se o pipeline está configurado para usar o metastore do Hive ou o Catálogo do Unity.

Metastore do Hive

Se o pipeline publicar tabelas no metastore do Hive, o log de eventos será armazenado em /system/events no local storage. Por exemplo, se você tiver definido a configuração storage de pipeline como /Users/username/data, o log de eventos será armazenado no caminho /Users/username/data/system/events no DBFS.

Se você não tiver definido a configuração storage, o local padrão do log de eventos será /pipelines/<pipeline-id>/system/events no DBFS. Por exemplo, se a ID do pipeline for 91de5e48-35ed-11ec-8d3d-0242ac130003, o local de armazenamento será /pipelines/91de5e48-35ed-11ec-8d3d-0242ac130003/system/events.

Você pode criar uma exibição para simplificar a consulta do log de eventos. O exemplo a seguir cria uma exibição temporária chamada event_log_raw. Essa exibição é usada nas consultas de log de eventos de exemplo incluídas neste artigo:

CREATE OR REPLACE TEMP VIEW event_log_raw AS SELECT * FROM delta.`<event-log-path>`;

Substitua <event-log-path> pelo local do log de eventos.

Cada instância de uma execução de pipeline é chamada de atualização. Geralmente, você deseja extrair informações para a atualização mais recente. Execute a consulta a seguir para encontrar o identificador da atualização mais recente e salve-o na exibição temporária latest_update_id. Essa exibição é usada nas consultas de log de eventos de exemplo incluídas neste artigo:

CREATE OR REPLACE TEMP VIEW latest_update AS SELECT origin.update_id AS id FROM event_log_raw WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1;

Você pode consultar o log de eventos em um notebook do Azure Databricks ou no editor SQL. Use um notebook ou o editor SQL para executar as consultas de log de eventos de exemplo.

Catálogo do Unity

Se o pipeline publicar tabelas no Catálogo do Unity, você deverá usar a event_logfunção com valor de tabela (TVF) para buscar o log de eventos do pipeline. Você recupera o log de eventos de um pipeline passando a ID do pipeline ou um nome de tabela para o TVF. Por exemplo, para recuperar os registros de log de eventos do pipeline com a ID 04c78631-3dd7-4856-b2a6-7d84e9b2638b:

SELECT * FROM event_log("04c78631-3dd7-4856-b2a6-7d84e9b2638b")

Para recuperar os registros de log de eventos do pipeline que criou ou possui a tabela my_catalog.my_schema.table1:

SELECT * FROM event_log(TABLE(my_catalog.my_schema.table1))

Para chamar o TVF, você deve usar um cluster compartilhado ou um SQL warehouse. Por exemplo, você pode usar um notebook anexado a um cluster compartilhado ou usar o editor SQL conectado a um SQL warehouse.

Para simplificar a consulta de eventos de um pipeline, o proprietário do pipeline pode criar uma exibição sobre o TVF event_log. O exemplo a seguir cria uma exibição sobre o log de eventos de um pipeline. Essa exibição é usada nas consultas de log de eventos de exemplo incluídas neste artigo.

Observação

O TVF event_log só pode ser chamado pelo proprietário do pipeline e uma exibição criada no TVF event_log só pode ser consultada pelo proprietário do pipeline. A exibição não pode ser compartilhada com outros usuários.

CREATE VIEW event_log_raw AS SELECT * FROM event_log("<pipeline-ID>");

Substitua <pipeline-ID> pelo identificador exclusivo do pipeline do Delta Live Tables. Você pode encontrar a ID no painel Detalhes do pipeline na interface do usuário do Delta Live Tables.

CREATE OR REPLACE TEMP VIEW latest_update AS SELECT origin.update_id AS id FROM event_log_raw WHERE event_type = 'create_update' ORDER BY timestamp DESC LIMIT 1;

Consultar informações de linhagem do log de evento

Os eventos que contêm informações sobre a linhagem têm o tipo de evento flow_definition. O objeto details:flow_definition contém o output_dataset e input_datasets definindo cada relação no grafo.

Você pode usar a consulta a seguir para extrair os conjuntos de dados de entrada e saída para ver as informações de linhagem:

SELECT
  details:flow_definition.output_dataset as output_dataset,
  details:flow_definition.input_datasets as input_dataset
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'flow_definition'
  AND
  origin.update_id = latest_update.id

	`output_dataset`	`input_datasets`
1	`customers`	`null`
2	`sales_orders_raw`	`null`
3	`sales_orders_cleaned`	`["customers", "sales_orders_raw"]`
4	`sales_order_in_la`	`["sales_orders_cleaned"]`

Consultar a qualidade dos dados do log de eventos

Se você definir expectativas sobre conjuntos de dados em seu pipeline, as métricas de qualidade de dados serão armazenadas no objeto details:flow_progress.data_quality.expectations. Os eventos que contêm informações sobre a qualidade de dados têm o tipo de evento flow_progress. O seguinte exemplo consulta as métricas de qualidade de dados em busca da última atualização do pipeline:

SELECT
  row_expectations.dataset as dataset,
  row_expectations.name as expectation,
  SUM(row_expectations.passed_records) as passing_records,
  SUM(row_expectations.failed_records) as failing_records
FROM
  (
    SELECT
      explode(
        from_json(
          details :flow_progress :data_quality :expectations,
          "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
        )
      ) row_expectations
    FROM
      event_log_raw,
      latest_update
    WHERE
      event_type = 'flow_progress'
      AND origin.update_id = latest_update.id
  )
GROUP BY
  row_expectations.dataset,
  row_expectations.name

	`dataset`	`expectation`	`passing_records`	`failing_records`
1	`sales_orders_cleaned`	`valid_order_number`	4083	0

Monitorar a lista de pendências de dados consultando o log de eventos

O Delta Live Tables rastreia quantos dados estão presentes na lista de pendências no objeto details:flow_progress.metrics.backlog_bytes. Os eventos que contêm as métricas de lista de pendências têm o tipo de evento flow_progress. O seguinte exemplo consulta as métricas de lista de pendências em busca da última atualização do pipeline:

SELECT
  timestamp,
  Double(details :flow_progress.metrics.backlog_bytes) as backlog
FROM
  event_log_raw,
  latest_update
WHERE
  event_type ='flow_progress'
  AND
  origin.update_id = latest_update.id

Observação

As métricas da lista de pendências podem não estar disponíveis dependendo do tipo de fonte de dados do pipeline e da versão do Databricks Runtime.

Monitorar os eventos de Dimensionamento Automático Aprimorado do log de eventos

O log de eventos captura redimensionamentos de cluster quando o Dimensionamento Automático Aprimorado está habilitado em seus pipelines. Os eventos que contêm informações sobre o Dimensionamento Automático Aprimorado têm o tipo de evento autoscale. As informações da solicitação de redimensionamento do cluster são armazenadas no objeto details:autoscale. O seguinte exemplo consulta as solicitações de redimensionamento do cluster de Dimensionamento Automático Aprimorado para a última atualização do pipeline:

SELECT
  timestamp,
  Double(
    case
      when details :autoscale.status = 'RESIZING' then details :autoscale.requested_num_executors
      else null
    end
  ) as starting_num_executors,
  Double(
    case
      when details :autoscale.status = 'SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'PARTIALLY_SUCCEEDED' then details :autoscale.requested_num_executors
      else null
    end
  ) as partially_succeeded_num_executors,
  Double(
    case
      when details :autoscale.status = 'FAILED' then details :autoscale.requested_num_executors
      else null
    end
  ) as failed_num_executors
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'autoscale'
  AND
  origin.update_id = latest_update.id

Monitorar a utilização dos recursos de computação

Os eventos cluster_resources fornecem métricas sobre o número de slots de tarefa no cluster, quanto se utiliza esses slots de tarefa e quantas tarefas estão aguardando para serem agendadas.

Quando o Dimensionamento Automático está habilitado, os eventos cluster_resources também contêm métricas para o algoritmo de dimensionamento automático, incluindo latest_requested_num_executors e optimal_num_executors. Os eventos também mostram o status do algoritmo como estados diferentes, como CLUSTER_AT_DESIRED_SIZE, SCALE_UP_IN_PROGRESS_WAITING_FOR_EXECUTORS e BLOCKED_FROM_SCALING_DOWN_BY_CONFIGURATION. Essas informações podem ser exibidas em conjunto com os eventos de dimensionamento automático para fornecer uma visão geral do Dimensionamento Automático Aprimorado.

O exemplo a seguir consulta o histórico de tamanho da fila de tarefas em busca da última atualização de pipeline:

SELECT
  timestamp,
  Double(details :cluster_resources.avg_num_queued_tasks) as queue_size
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

O exemplo a seguir consulta o histórico de utilização em busca da última atualização de pipeline:

SELECT
  timestamp,
  Double(details :cluster_resources.avg_task_slot_utilization) as utilization
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

O exemplo a seguir consulta o histórico de contagem do executor, acompanhado de métricas disponíveis apenas para pipelines de Dimensionamento Automático Aprimorado, incluindo o número de executores solicitados pelo algoritmo na solicitação mais recente, o número ideal de executores recomendado pelo algoritmo com base nas métricas mais recentes e o estado do algoritmo de dimensionamento automático:

SELECT
  timestamp,
  Double(details :cluster_resources.num_executors) as current_executors,
  Double(details :cluster_resources.latest_requested_num_executors) as latest_requested_num_executors,
  Double(details :cluster_resources.optimal_num_executors) as optimal_num_executors,
  details :cluster_resources.state as autoscaling_state
FROM
  event_log_raw,
  latest_update
WHERE
  event_type = 'cluster_resources'
  AND
  origin.update_id = latest_update.id

Auditoria de pipelines do Delta Live Tables

Você pode usar os registros de log de eventos do Delta Live Tables e outros logs de auditoria do Azure Databricks para obter uma imagem completa de como os dados estão sendo atualizados no Delta Live Tables.

O Delta Live Tables usa as credenciais do proprietário do pipeline para executar atualizações. Você pode alterar as credenciais usadas atualizando o proprietário do pipeline. O Delta Live Tables registra o usuário para ações no pipeline, incluindo a criação do pipeline, edições na configuração e o disparo de atualizações.

Confira Eventos do Catálogo do Unity para obter uma referência de eventos de auditoria do Catálogo do Unity.

Use o log de eventos para auditar eventos, por exemplo, as ações do usuário. Os eventos que contêm informações sobre as ações do usuário têm o tipo de evento user_action.

As informações sobre a ação são armazenadas no objeto user_action no campo details. Use a seguinte consulta para construir um log de auditoria dos eventos de usuário. Para criar a exibição event_log_raw usada nesta consulta, confira Consultando o log de eventos.

SELECT timestamp, details:user_action:action, details:user_action:user_name FROM event_log_raw WHERE event_type = 'user_action'

	`timestamp`	`action`	`user_name`
1	2021-05-20T19:36:03.517+0000	`START`	`user@company.com`
2	2021-05-20T19:35:59.913+0000	`CREATE`	`user@company.com`
3	2021-05-27T00:35:51.971+0000	`START`	`user@company.com`

Informações de runtime

Você pode exibir informações de runtime para uma atualização de pipeline, por exemplo, a versão do Databricks Runtime para a atualização:

SELECT details:create_update:runtime_version:dbr_version FROM event_log_raw WHERE event_type = 'create_update'

	`dbr_version`
1	11.0

Share via