Apresentando a API DLT Sink: Escreva Pipelines para Kafka e Tabelas Delta Externas

Se você é novo no Delta Live Tables, antes de ler este blog, recomendamos a leitura de Introdução ao Delta Live Tables, que explica como você pode criar pipelines escaláveis e confiáveis usando as definições e declarações de ETL declarativas do Delta Live Tables (DLT).

Introdução

Delta Live Tables (DLT) oferecem uma plataforma robusta para a construção de pipelines de processamento de dados confiáveis, sustentáveis e testáveis dentro do Databricks. Ao aproveitar seu framework declarativo e provisionar automaticamente o cálculo serverless ideal, o DLT simplifica as complexidades de streaming, transformação de dados e gerenciamento, proporcionando escalabilidade e eficiência para fluxos de trabalho de dados modernos.

Tradicionalmente, as Pipelines DLT ofereceram uma maneira eficiente de ingerir e processar dados como Tabelas de Streaming ou Visualizações Materializadas governadas pelo Catálogo Unity. Embora essa abordagem atenda à maioria das necessidades de processamento de dados, existem casos em que os pipelines de dados devem se conectar com sistemas externos ou precisam usar sinks de Streaming Estruturado em vez de escrever em Tabelas de Streaming ou Visualizações Materializadas.

A introdução da nova API Sinks no DLT aborda isso, permitindo que os usuários escrevam dados processados em fluxos de eventos externos, como Apache Kafka, Azure Event Hubs, bem como escrevendo em uma Tabela Delta. Essa nova capacidade amplia o escopo dos pipelines DLT, permitindo uma integração perfeita com plataformas externas.

Essas funcionalidades estão agora em Visualização Pública e continuaremos a adicionar mais destinos do Databricks Runtime para DLT ao longo do tempo, eventualmente suportando todos eles. O próximo em que estamos trabalhando é foreachBatch que permite aos clientes escrever em qualquer destino de dados e realizar mesclagens personalizadas em tabelas Delta.

A API Sink está disponível no pacote Python dlt e pode ser usada com create_sink() como mostrado abaixo:

A API aceita três argumentos-chave na definição do sink:

• Nome do Sink: Uma string que identifica unicamente o sink dentro do seu pipeline. Este nome permite que você faça referência e gerencie o destino.
• Especificação de Formato: Uma string que determina o formato de saída, com suporte para "kafka" ou "delta".
• Opções do Sink: Um dicionário de pares chave-valor, onde ambas as chaves e valores são strings. Para os sinks do Kafka, todas as opções de configuração disponíveis no Structured Streaming podem ser aproveitadas, incluindo configurações para autenticação, estratégias de particionamento e mais. Por favor, consulte a documentação para uma lista completa das opções de configuração suportadas pelo Kafka. Os destinos Delta oferecem uma configuração mais simples, permitindo que você defina um caminho de armazenamento usando o atributo path ou escreva diretamente em uma tabela no Unity Catalog usando o atributo tableName.

Escrevendo para um Sink

A API @append_flow foi aprimorada para permitir a escrita de dados em sinks alvo identificados por seus nomes de sink. Tradicionalmente, esta API permitia aos usuários carregar dados de várias fontes em uma única tabela de streaming de forma contínua. Com a nova melhoria, os usuários agora podem também anexar dados a sinks específicos. Abaixo está um exemplo de como configurar isso:

Construindo o pipeline

Vamos agora construir uma pipeline DLT que processa dados de clickstream, embalados nos conjuntos de dados Databricks. Esta pipeline irá analisar os dados para identificar eventos que levam a uma página Apache Spark e, posteriormente, escrever esses dados tanto nos Event Hubs quanto nos sinks Delta. Vamos estruturar a pipeline usando a Arquitetura Medallion, que organiza os dados em diferentes camadas para melhorar a qualidade e a eficiência do processamento.

Começamos carregando dados JSON brutos na camada Bronze usando o Auto Loader. Em seguida, limpamos os dados e aplicamos padrões de qualidade na camada Silver para garantir sua integridade. Finalmente, na camada Gold, filtramos as entradas com um título de página atual de Apache_Spark e as armazenamos em uma tabela chamada spark_referrers, que servirá como a fonte para nossos sinks. Por favor, consulte o Apêndice para o código completo.

Configurando o Sink Azure Event Hubs

Nesta seção, usaremos a API create_sink para estabelecer um sink Event Hubs. Isso pressupõe que você tenha um fluxo operacional Kafka ou Event Hubs. Nosso pipeline transmitirá dados para Event Hubs habilitados para Kafka usando uma política de acesso compartilhado, com a string de conexão armazenada de forma segura nos Segredos Databricks. Alternativamente, você pode usar um principal de serviço para integração em vez de uma política SAS. Certifique-se de atualizar as propriedades de conexão e segredos de acordo. Aqui está o código para configurar o destino Event Hubs:

Configurando o Delta Sink

Além do destino Event Hubs, podemos utilizar a API create_sink para configurar um destino Delta. Este sink escreve dados em um local especificado no Sistema de Arquivos Databricks (DBFS), mas também pode ser configurado para escrever em um local de armazenamento de objetos, como Amazon S3 ou ADLS.

Abaixo está um exemplo de como configurar um sink Delta:

Criando Fluxos para hidratar sinks Kafka e Delta

Com os sinks Event Hubs e Delta estabelecidos, o próximo passo é hidratar esses sinks usando o decorador append_flow. Esse processo envolve o streaming de dados para os sinks, garantindo que eles sejam continuamente atualizados com as informações mais recentes.

Para o sink do Event Hubs, o parâmetro value é obrigatório, enquanto parâmetros adicionais como key, partition, headers e topic podem ser especificados opcionalmente. Abaixo estão exemplos de como configurar fluxos para os sinks do Kafka e Delta:

A função applyInPandasWithState também agora é suportada em DLT, permitindo aos usuários aproveitar o poder do Pandas para processamento com estado em seus pipelines DLT. Este aprimoramento permite transformações de dados mais complexas e agregações usando a familiar API Pandas. Com a API DLT Sink, os usuários podem facilmente transmitir esses dados processados ​​com estado para tópicos Kafka. Esta integração é particularmente útil para análises em tempo real e arquiteturas orientadas a eventos, garantindo que os pipelines de dados possam lidar eficientemente e distribuir dados de streaming para sistemas externos.

Juntando tudo

A abordagem demonstrada acima mostra como construir uma pipeline DLT que transforma eficientemente os dados enquanto utiliza a nova API Sink para entregar os resultados de forma contínua para Tabelas Delta externas e Event Hubs habilitados para Kafka.

Este recurso é particularmente valioso para pipelines de análise em tempo real, permitindo que os dados sejam transmitidos para streams Kafka para aplicações como detecção de anomalias, manutenção preditiva e outros casos de uso sensíveis ao tempo. Ele também permite arquiteturas orientadas a eventos, onde processos downstream podem ser acionados instantaneamente por eventos de streaming para tópicos Kafka, permitindo o rápido processamento de dados recém-chegados.

Chamada para Ação

O recurso DLT Sinks agora está disponível em Visualização Pública para todos os clientes Databricks! Esta nova capacidade poderosa permite que você estenda seus pipelines DLT para sistemas externos como Kafka e tabelas Delta, garantindo fluxo de dados em tempo real e integrações simplificadas. Para mais informações, por favor, consulte os seguintes recursos:

• Documentação do Databricks sobre Sinks DLT
• Assista a demonstração sobre Kafka Sinks em DLT usando Confluent Cloud
• Referência da Linguagem Python para DLT

Apêndice:

Código do Pipeline:

(This blog post has been translated using AI-powered tools) Original Post