O que o destinatário realmente faz com uma tabela compartilhada
por Jordan Soldo e Nikhil Tallam
Muitos arquiteturas de compartilhamento de dados começam com um padrão familiar:
A Parte 1 deste blog descreveu por que uma tabela compartilhada pode ser uma alternativa melhor do que um feed de arquivos recorrente ou de API. Agora, vamos ver o que o destinatário realmente faz com essa tabela.
Este passo a passo foca no compartilhamento entre Databricks e Databricks (D2D). O destinatário tem um workspace do Databricks com o Unity Catalog ativado e acessa a tabela compartilhada por meio de um catálogo compartilhado. O provedor compartilha a tabela com seu histórico, permitindo que o destinatário a use como uma fonte do Structured Streaming. Voltaremos a falar sobre o compartilhamento entre Databricks e Open (D2O) mais adiante na publicação.
Uma Delta table mantém um histórico ordenado de cada alteração feita nela. O OpenSharing expõe essa tabela por meio de um protocolo aberto sem copiá-la. Os dados permanecem no armazenamento do provedor, e o destinatário os lê em tempo real.
No lado do destinatário, você aponta um leitor do Structured Streaming para a tabela compartilhada. O Spark registra até onde ele leu em um checkpoint, de modo que cada execução coleta apenas as alterações feitas desde a última. Rastrear "o que há de novo desde a última vez que verifiquei" passa a ser tarefa do engine, executando sobre dados governados com linhagem e controle de acesso intactos.
Vamos usar um exemplo: um varejista mantém uma tabela de pedidos em tempo real e a compartilha com um destinatário externo que executa o Databricks. Em vez de receber um CSV todas as noites, o destinatário pode consultar a tabela de pedidos compartilhada diretamente. Se o destinatário precisar de uma cópia local para análises ou cargas de trabalho posteriores, ele poderá materializar os dados incrementalmente em uma Delta table. A configuração é feita uma única vez e é dividida entre o provedor e o destinatário.
Crie a tabela de origem com o Change Data Feed ativado e, em seguida, publique-a em um compartilhamento COM HISTÓRICO (WITH HISTORY). O histórico é o que torna a tabela uma fonte incremental. Sem ele, o destinatário obtém apenas um snapshot estático. Ativar o Change Data Feed (delta.enableChangeDataFeed) é opcional para a leitura incremental principal abaixo. Ative-o apenas se você planeja usar o caminho de upserts e exclusões mais tarde. Se você estiver no Databricks Runtime 19 ou superior, poderá ignorar a propriedade e contar com o Change Data Feed automático, que calcula alterações em nível de linha no momento da consulta para tabelas com rastreamento de linhas ativado e funciona com o OpenSharing de Databricks para Databricks.
Registre o destinatário pelo identificador de compartilhamento e, em seguida, conceda o compartilhamento:
Observe que CREATE SHARE e CREATE RECIPIENT exigem privilégio no nível do metastore, e não apenas de administrador do workspace.
O compartilhamento agora aparece como um provedor no seu metastore. Monte-o como um catálogo e você poderá lê-lo como qualquer outra tabela.
Com o compartilhamento montado como shared_in, a leitura é tratada pelo Structured Streaming.
Trigger.AvailableNow processa todos os dados disponíveis e não processados usando um ou mais micro-lotes e, em seguida, é encerrado. Isso o torna ideal para jobs incrementais agendados. Em vez de um agendamento fixo, você também pode configurar o job com um gatilho de atualização de tabela na tabela compartilhada, para que o job seja executado somente quando o provedor confirmar novos dados. É recomendável armazenar o checkpoint em um armazenamento em nuvem durável, como UC Volumes.
Observação: este exemplo lê uma tabela compartilhada por meio do Delta Sharing, portanto, ele usa ignoreChanges. Se você fizer streaming diretamente de uma Delta table normal, os runtimes mais recentes do Databricks usarão skipChangeCommits para processamento exclusivo de anexação (append-only).
Para provar que é realmente incremental, execute o stream uma vez em um conjunto inicial de linhas. Execute-o novamente depois de ter novas linhas no provedor. O histórico de confirmação da tabela de destino mostrará duas gravações de streaming:
Version | Operation | Operation Metrics |
|---|---|---|
2 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 200, numOutputBytes: 2307, numAddedFiles: 1 |
1 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 1000, numOutputBytes: 4958, numAddedFiles: 1 |
0 | CREATE TABLE | {} |
A segunda execução gravou 200 linhas, não 1200. O checkpoint rastreou a última versão lida, portanto, a execução puxou apenas as novas.
ignoreChanges é adequado para esse padrão de chegada incremental, mas pode reemitir linhas e não propaga exclusões. Se os consumidores downstream precisarem de inserções, atualizações e exclusões, leia o Change Data Feed da tabela compartilhada:
Quando um novo cliente é adicionado, o provedor não precisa criar um novo endpoint SFTP nem implementar um novo job de exportação. O provedor concede ao destinatário acesso ao compartilhamento. O destinatário usa o mesmo padrão geral, mantendo suas próprias tabelas de destino e lógica de negócios downstream separadas. A diferença operacional é significativa:
Tudo o que está acima funciona porque o destinatário executa o Spark no Databricks, o que desbloqueia o streaming com checkpoint e o CDF. Outros destinatários (pandas, ferramentas de BI, Snowflake) se conectam por meio do OpenSharing com um arquivo de credenciais em vez de um ID de compartilhamento de metastore.
Se eles executarem o OSS Spark, a leitura será o mesmo código spark.readStream por meio do conector delta-sharing. Sem o Spark, não há stream contínuo, portanto, eles puxam as alterações por intervalo de versões:
Um primeiro projeto útil não precisa resolver todos os cenários de compartilhamento. Comece com uma Delta table normal e um destinatário.
Antes de implementar o stream, responda a estas perguntas:
Assim que essas respostas estiverem claras, valide o menor fluxo ponta a ponta possível: crie o compartilhamento, leia um snapshot inicial, grave uma nova linha, atualize uma linha, exclua uma linha e confirme se o destinatário vê os resultados esperados.
Essa é a mudança prática que o OpenSharing torna possível. Em vez de criar um novo fluxo de trabalho de transporte e ingestão para cada cliente, você pode estabelecer um contrato de compartilhamento de dados governado e permitir que cada destinatário consuma os dados por meio do caminho que melhor se adapta à sua plataforma.
Novo no OpenSharing? Comece com O que é o OpenSharing? e Criar compartilhamentos para o OpenSharing, depois configure o fluxo básico de ponta a ponta acima usando uma tabela e um destinatário.
(Esta publicação no blog foi traduzida utilizando ferramentas baseadas em inteligência artificial) Publicação original
Assine nosso blog e receba os posts mais recentes diretamente na sua caixa de entrada.