Apresentando a API de Sink DLT: 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 definições e instruções ETL declarativas do Delta Live Tables (DLT).

Introdução

Delta Live Tables (DLT) pipelines oferecem uma plataforma robusta para construir pipelines de processamento de dados confiáveis, fáceis de manter e testáveis dentro do Databricks. Ao aproveitar sua estrutura declarativa e provisionar automaticamente computação serverless otimizada, o DLT simplifica as complexidades de streaming, transformação e gerenciamento de dados, entregando escalabilidade e eficiência para fluxos de trabalho de dados modernos.

Tradicionalmente, os pipelines DLT ofereciam uma maneira eficiente de ingerir e processar dados como Streaming Tables ou Materialized Views, governados pelo Unity Catalog. Embora essa abordagem atenda à maioria das necessidades de processamento de dados, existem casos em que os pipelines de dados precisam se conectar a sistemas externos ou usar sinks do Structured Streaming em vez de escrever em Streaming Tables ou Materialized Views.

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

Esses recursos agora estão em Pré-visualização Pública e continuaremos a adicionar mais sinks do Databricks Runtime ao DLT ao longo do tempo, eventualmente suportando todos eles. O próximo que estamos desenvolvendo é o foreachBatch, que permite aos clientes escrever em sinks de dados arbitrários e realizar merges customizados em Delta Tables.

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

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

• Nome do Sink: Uma string que identifica unicamente o sink dentro do seu pipeline. Este nome permite que você referencie e gerencie o sink.
• 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 tanto as chaves quanto os valores são strings. Para sinks Kafka, todas as opções de configuração disponíveis no Structured Streaming podem ser utilizadas, incluindo configurações para autenticação, estratégias de particionamento e mais. Por favor, consulte a documentação para uma lista completa de opções de configuração suportadas pelo Kafka. Sinks 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 de destino identificados por seus nomes. Tradicionalmente, esta API permitia que os usuários carregassem dados de várias fontes para uma única streaming table sem problemas. Com o novo aprimoramento, os usuários agora também podem adicionar dados a sinks específicos. Abaixo está um exemplo demonstrando como configurar isso:

Construindo o Pipeline

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

Começamos carregando dados brutos JSON 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 fonte para nossos sinks. Consulte o Apêndice para o código completo.

Configurando o Sink do Azure Event Hubs

Nesta seção, usaremos a API create_sink para estabelecer um sink Event Hubs. Isso pressupõe que você tenha um stream Kafka ou Event Hubs operacional. Nosso pipeline transmitirá dados para Event Hubs habilitados para Kafka usando uma política de acesso compartilhado, com a string de conexão armazenada com segurança no Databricks Secrets. 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 os segredos de acordo. Aqui está o código para configurar o sink Event Hubs:

Configurando o Sink Delta

Além do sink Event Hubs, podemos usar a API create_sink para configurar um sink Delta. Este sink grava dados em um local especificado no Databricks File System (DBFS), mas também pode ser configurado para gravar 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 Popular os Sinks Kafka e Delta

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

Para o sink 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 Kafka e Delta:

A função applyInPandasWithState também é agora suportada no DLT, permitindo que os usuários aproveitem o poder do Pandas para processamento com estado em seus pipelines DLT. Este aprimoramento permite transformações e agregações de dados mais complexas usando a API Pandas familiar. 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 e distribuir dados de streaming de forma eficiente para sistemas externos.

Juntando Tudo

A abordagem demonstrada acima mostra como construir um pipeline DLT que transforma dados de forma eficiente, utilizando a nova API Sink para entregar os resultados de forma integrada a 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 fluxos Kafka para aplicações como detecção de anomalias, manutenção preditiva e outros casos de uso sensíveis ao tempo. Ele também habilita arquiteturas orientadas a eventos, onde processos downstream podem ser acionados instantaneamente pela transmissão de eventos para tópicos Kafka, permitindo o processamento rápido de dados recém-chegados.

Chamada para Ação

O recurso DLT Sinks agora está disponível em Public Preview para todos os clientes Databricks! Essa nova e poderosa funcionalidade permite estender seus pipelines DLT de forma integrada a sistemas externos como Kafka e tabelas Delta, garantindo fluxo de dados em tempo real e integrações simplificadas. Para mais informações, consulte os seguintes recursos:

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

Apêndice:

Código do Pipeline:

(Esta publicação no blog foi traduzida utilizando ferramentas baseadas em inteligência artificial) Publicação original