Pular para o conteúdo principal
Última atualização em

Usar tabelas de transmissão autônomas

Uma tabela de transmissão independente é uma tabela registrada no Unity Catalog com suporte extra para transmissão ou processamento incremental de dados, definido fora de um pipeline Lakeflow. Um pipeline é criado automaticamente para cada tabela de transmissão. Você pode usar tabelas de transmissão para carregamento incremental de dados do Kafka e armazenamento de objetos na cloud.

Você pode criar e refresh tabelas de streaming autônomas a partir de um Databricks SQL warehouse ou de um Notebook executado em compute geral serverless. Para obter detalhes sobre as diferenças entre as duas opções de compute, consulte Requisitos para pipelines autônomos.

Para criar e refresh tabelas de transmissão autônomas com Python a partir de um notebook, consulte Usar Python com pipelines autônomos.

nota

Para aprender a usar tabelas Delta Lake como fontes e coletores de transmissão, consulte Leituras e gravações de transmissão de tabela Delta Lake.

Requisitos​

Para opções de compute, permissões e outros requisitos para criar, atualizar e consultar tabelas de transmissão autônomas, consulte Requisitos para pipelines autônomos.

Criar tabelas de transmissão​

Uma tabela de transmissão é definida por uma query SQL no Databricks SQL. Quando você cria uma tabela de transmissão, os dados atualmente nas tabelas de origem são usados para criar a tabela de transmissão. Depois disso, você refresh a tabela, geralmente em um programar, para inserir quaisquer dados adicionais nas tabelas de origem para anexar à tabela de transmissão.

Ao criar uma tabela de transmissão, você é considerado o proprietário da tabela.

Para criar uma tabela de transmissão a partir de uma tabela existente, utilize a instrução CREATE STREAMING TABLE, como no seguinte exemplo:

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT product, price FROM STREAM raw_data;

Nesse caso, a tabela de transmissão sales é criada a partir de colunas específicas da tabela raw_data, com uma programação para realizar o refresh a cada hora. A query usada deve ser uma query de transmissão . Use a palavra-chave STREAM para usar semântica de transmissão para ler da fonte.

Compute usado para refresh​

Quando você cria uma tabela de transmissão usando a declaração CREATE OR REFRESH STREAMING TABLE, o refresh inicial dos dados e o preenchimento começam imediatamente. Essas operações não consomem compute do SQL Warehouse do Databricks. Em vez disso, as tabelas de transmissão dependem de pipelines Serverless para criação e refresh. Um pipeline serverless dedicado é criado e gerenciado automaticamente pelo sistema para cada tabela de transmissão.

Carregar arquivos com o Auto Loader​

Para criar uma tabela de transmissão a partir de arquivos em um volume, você usa o Auto Loader. Use o Auto Loader para a maioria das tarefas de ingestão de dados a partir do armazenamento de objetos na cloud. O Auto Loader e os pipelines são projetados para carregar de forma progressiva e idempotente dados que aumentam constantemente à medida que chegam ao armazenamento em cloud.

Para usar o Auto Loader no Databricks SQL, use a função read_files. Os exemplos a seguir mostram o uso do Auto Loader para ler um volume de arquivos JSON em uma tabela de transmissão:

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT * FROM STREAM read_files(
    "/Volumes/my_catalog/my_schema/my_volume/path/to/data",
    format => "json"
  );

Para ler dados do armazenamento em nuvem, você também pode usar o Auto Loader:

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
  's3://mybucket/analysis/*/*/*.json',
    format => "json"
  );

Para saber mais sobre o Auto Loader, consulte O que é o Auto Loader?. Para saber mais sobre como usar o Auto Loader em SQL, com exemplos, consulte Carregar dados do armazenamento de objetos.

Ingestão de transmissão de outras fontes​

Para ver um exemplo de ingestão de outras fontes, incluindo Kafka, consulte Carregar dados em pipelines.

Aplicar captura de dados de alterações (CDC) com fluxos CDC automáticos​

Use a cláusula FLOW AUTO CDC para processar registros de captura de dados de alterações (CDC) de uma fonte em uma tabela de transmissão. Anteriormente, a declaração MERGE INTO era comumente usada para processar registros CDC no Databricks. No entanto, MERGE INTO pode produzir resultados incorretos devido a registros fora de sequência ou exige uma lógica complexa para reordenar os registros. Consulte captura de dados de alterações (CDC) e Snapshots.

AUTO CDC simplifica o CDC ao lidar automaticamente com registros fora de ordem. Você especifica keys para identificar registros, uma coluna de sequência para ordenação e se deseja armazenar os resultados como SCD tipo 1 (atualizações diretas) ou SCD tipo 2 (acompanhamento de história).

O seguinte exemplo cria uma tabela de transmissão que aplica alterações de CDC usando SCD tipo 1:

SQL
CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

O exemplo a seguir usa SCD tipo 2 para manter uma história de alterações:

SQL
CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2;

Para obter detalhes completos sobre opções e comportamento do Auto CDC, consulte As APIs AUTO CDC: Simplifique a captura de dados de alterações (CDC) com pipelines. Para obter a referência completa da sintaxe, consulte CREATE STREAMING TABLE.

Aplicar substituição de lote seletiva com fluxos REPLACE WHERE​

Use a cláusula FLOW REPLACE WHERE para recalcular e sobrescrever um subconjunto de destino de uma tabela de transmissão sem processar toda a história da tabela. Os fluxos REPLACE WHERE são adequados para o processamento em lotes incremental de joins e agregações, dados que chegam com atraso, reprocessamento upstream, evolução do esquema e backfills.

Para obter detalhes completos sobre fluxos REPLACE WHERE, incluindo requisitos, substituições de predicado e refresh incremental, consulte Fluxos REPLACE WHERE para tabelas de transmissão autônomas.

Aplicar substituição parcial de snapshot com fluxos REPLACE USING​

info

Beta

Fluxos de REPLACE USING estão em Beta.

Use a cláusula FLOW REPLACE USING para manter uma tabela de transmissão sincronizada com uma transmissão de Snapshots parciais. Em cada atualização, um fluxo REPLACE USING substitui todas as linhas que correspondem às colunas de key especificadas e deixa todas as outras linhas inalteradas. Uma coluna SEQUENCE BY ordena as atualizações para que a sequência mais alta de uma key vença sempre, mesmo quando as atualizações chegarem fora de ordem. Por exemplo:

SQL
CREATE OR REFRESH STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

BY NAME é obrigatório. Ele corresponde as colunas por nome em vez de posição.

O SUBSTITUIR USANDO se comporta da mesma forma para tabelas de transmissão autônomas e em pipelines Lakeflow. Para saber como funciona, sequenciamento, expectativas, limitações e exemplos, consulte Substituição parcial de snapshot com fluxos REPLACE USING. As seguintes diferenças se aplicam a tabelas de transmissão autônomas:

  • Defina o fluxo em SQL. Crie o fluxo REPLACE USING com a cláusula SQL FLOW REPLACE USING inline em CREATE OR REFRESH STREAMING TABLE. A instrução CREATE FLOW autônoma é uma construção de LakeFlow Pipelines e não é usada para tabelas de transmissão autônomas.
  • O compute é gerenciado para você. Tabelas de transmissão independentes estão em execução em pipelines Serverless gerenciados pelo sistema e exigem o Databricks Runtime 18.2 ou acima. Você não escolhe entre compute clássico e serverless.

Ingerir apenas novos dados​

Por default, a função read_files lê todos os dados existentes na pasta de origem durante a criação da tabela e, em seguida, processa os registros recém-chegados com cada refresh.

Para evitar a ingestão de dados que já existem na pasta de origem no momento da criação da tabela, defina a opção includeExistingFiles como false. Isso significa que apenas os dados que chegam à pasta após a criação da tabela são processados. Por exemplo:

SQL
CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    '/path/to/files',
    includeExistingFiles => false
  );

Versão do Runtime​

As tabelas de transmissão sempre são executadas na versão mais recente do tempo de execução do Databricks SQL. A propriedade de tabela pipelines.channel, usada anteriormente para selecionar um canal de tempo de execução preview ou current, não é mais compatível e não tem efeito. Se uma definição existente incluir essa propriedade, ela será ignorada com segurança e não será necessário removê-la.

Ocultar dados confidenciais​

Você pode usar tabelas de transmissão para ocultar dados sensíveis de usuários que acessam a tabela. Uma abordagem é definir a query para que ela exclua completamente as colunas ou linhas sensíveis. Alternativamente, você pode aplicar máscaras de coluna ou filtros de linha com base nas permissões do usuário que está fazendo a consulta. Por exemplo, você pode ocultar a coluna tax_id para usuários que não estão no grupo HumanResourcesDept. Para fazer isso, use a sintaxe ROW FILTER e MASK durante a criação da tabela de transmissão. Para obter mais informações, consulte Filtros de linha e máscaras de coluna.

refresh uma tabela de transmissão​

As tabelas de transmissão criam e usam automaticamente pipelines serverless para processar operações de refresh. O refresh é gerenciado pelo pipeline e a atualização é monitorada pelo warehouse Databricks SQL usado para criar a tabela de transmissão. As tabelas de transmissão podem ser atualizadas usando um pipeline que é executado em um programar.

Mesmo que você tenha um refresh programado, você pode fazer um refresh manual a qualquer momento. Os refresh são gerenciados pelo mesmo pipeline que foi criado automaticamente junto com a tabela de streaming.

Para fazer refresh de uma tabela de transmissão:

SQL
REFRESH STREAMING TABLE sales;

Você pode verificar o status do último refresh com DESCRIBE TABLE EXTENDED.

nota

Pode ser necessário fazer refresh da sua tabela de transmissão antes de usar queries de viagem do tempo.

Para saber como programar uma refresh, consulte Programar refresh. Os refresh agendados podem ter notificações de atualização, e você pode definir o modo de desempenho para o refresh.

Como funciona o refresh​

Um refresh de tabela de transmissão avalia apenas novas linhas que chegaram após a última atualização e acrescenta apenas os novos dados.

Cada refresh usa a definição atual da tabela de transmissão para processar esses novos dados. Modificar uma definição de tabela de transmissão não recalcula automaticamente os dados existentes. Se uma modificação for incompatível com os dados existentes (por exemplo, alterar um tipo de dados), o próximo refresh falhará com um erro.

Os exemplos a seguir explicam como as alterações em uma definição de tabela de transmissão afetam o comportamento de refresh:

  • A remoção de um filtro não reprocessa linhas filtradas anteriormente.
  • Alterar as projeções de colunas não afeta a forma como os dados existentes foram processados.
  • Os joins com snapshots estáticos usam o estado do snapshot no momento do processamento inicial. Dados que chegam atrasados e que teriam correspondido ao snapshot atualizado são ignorados. Isso pode levar ao descarte de fatos se as dimensões chegarem atrasadas.
  • A modificação da conversão de uma coluna existente resulta em um erro.

Se seus dados mudarem de uma forma que não possa ser suportada na tabela de transmissão existente, você poderá realizar uma refresh completa.

refresh totalmente uma tabela de transmissão​

Os refresh completos reprocessam todos os dados disponíveis na origem com a definição mais recente. Não é recomendado chamar refresh completos em fontes que não mantêm toda a história dos dados ou têm períodos de retenção curtos, como o Kafka, porque o refresh completo trunca os dados existentes. Talvez você não consiga recuperar dados antigos se os dados não estiverem mais disponíveis na origem.

Por exemplo:

SQL
REFRESH STREAMING TABLE sales FULL;

Programar e monitorar refresh​

Você pode refresh uma tabela de transmissão automaticamente em uma programar ou quando os dados upstream forem alterados, e você pode configurar tempos limite de refresh, notificações e modos de desempenho. Consulte programar refresh.

Controle o acesso a tabelas de transmissão​

As tabelas de transmissão oferecem suporte a controles de acesso avançados para oferecer suporte ao compartilhamento de dados enquanto evita a exposição de dados potencialmente privados. Um proprietário de tabela de transmissão ou um usuário com o privilégio MANAGE pode conceder privilégios SELECT a outros usuários. Usuários com acesso SELECT à tabela de transmissão não exigem acesso SELECT às tabelas referenciadas pela tabela de transmissão. Esse controle de acesso permite o compartilhamento de dados enquanto controla o acesso aos dados subjacentes.

Você também pode modificar o proprietário de uma tabela de transmissão.

Conceder privilégios a uma tabela de transmissão​

Para conceder acesso a uma tabela de transmissão, use a instrução GRANT:

SQL
GRANT <privilege_type> ON <st_name> TO <principal>;

O privilege_type pode ser:

  • SELECT - o usuário pode SELECT a tabela de transmissão.
  • REFRESH - o usuário pode REFRESH a tabela de transmissão. Os refresh são executados usando as permissões do proprietário.

O exemplo a seguir cria uma tabela de transmissão e concede privilégios de seleção e refresh aos usuários:

SQL
CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM STREAM source_table;

-- Grant read-only access:
GRANT SELECT ON st_name TO read_only_user;

-- Grant read and refresh access:
GRANT SELECT ON st_name TO refresh_user;
GRANT REFRESH ON st_name TO refresh_user;

Para mais informações sobre como conceder privilégios em objetos seguros do Unity Catalog, consulte referência de privilégios do Unity Catalog.

Revogar privilégios de uma tabela de transmissão​

Para revogar o acesso de uma tabela de transmissão, use a instrução REVOKE:

SQL
REVOKE privilege_type ON <st_name> FROM principal;

Quando os privilégios de SELECT em uma tabela de origem são revogados do proprietário da tabela de transmissão ou de qualquer outro usuário que tenha recebido privilégios de MANAGE ou SELECT na tabela de transmissão, ou quando a tabela de origem é excluída, o proprietário da tabela de transmissão ou o usuário que recebeu acesso ainda consegue fazer query na tabela de transmissão. No entanto, ocorre o seguinte comportamento:

  • O proprietário da tabela de transmissão ou outros que perderam o acesso a uma tabela de transmissão não podem mais REFRESH essa tabela de transmissão, e a tabela de transmissão torna-se obsoleta com o passar do tempo.
  • Se automatizado com programar, a próxima execução REFRESH programada falhará ou não será executada.

O exemplo a seguir revoga o privilégio SELECT de read_only_user:

SQL
REVOKE SELECT ON st_name FROM read_only_user;

Alterar o proprietário de uma tabela de transmissão​

Um usuário com as permissões MANAGE em uma tabela de transmissão autônoma pode definir um novo proprietário através do Explorador de Catálogos. O novo proprietário pode ser ele mesmo ou uma Service Principal em que ele tenha a função de Usuário Service Principal .

  1. No seu workspace do Databricks, clique em Ícone de dados. Catálogo para abrir o Catalog Explorer.

  2. Selecione a tabela de transmissão que você deseja atualizar.

  3. Na barra lateral direita, em Sobre esta tabela de transmissão , localize o Proprietário e clique em Ícone de lápis. editar.

nota

Se você receber uma mensagem que diz para atualizar o proprietário alterando o usuário Executar como nas configurações do pipeline, a tabela de transmissão será definida em um pipeline do Lakeflow, não como uma tabela autônoma. A mensagem inclui um link para as configurações do pipeline, onde você pode alterar o usuário execução como .

  1. Selecione um novo proprietário para a tabela de transmissão.

    Os proprietários recebem automaticamente privilégios de MANAGE e SELECT sobre tabelas de transmissão que eles possuem. Se você estiver definindo um service principal como proprietário de uma tabela de transmissão que você possui, e você não tiver explicitamente privilégios de SELECT ou MANAGE na tabela de transmissão, essa alteração fará com que você perca todo o acesso à tabela de transmissão. Nesse caso, você será solicitado a fornecer explicitamente esses privilégios.

    Selecione os privilégios Conceder GERENCIAR e Conceder SELECIONAR para fornecê-los ao Salvar .

  2. Clique em Salvar para alterar o proprietário.

O proprietário da tabela de transmissão será atualizado. Todas as atualizações futuras serão executadas usando a identidade do novo proprietário.

Quando o proprietário perde privilégios para tabelas de origem​

Se você alterar o proprietário e o novo proprietário não tiver acesso às tabelas de origem (ou se os privilégios SELECT forem revogados nas tabelas de origem subjacentes), os usuários ainda poderão fazer query na tabela de transmissão. No entanto:

  • Eles não podem REFRESH a tabela de transmissão.
  • A próxima refresh agendada da tabela de transmissão falha.

Perder o acesso aos dados de origem impede atualizações, mas não invalida imediatamente a tabela de transmissão existente de ser lida.

Excluir registros permanentemente de uma tabela de transmissão​

info

Visualização

O suporte para a declaração REORG com tabelas de transmissão está em Pré-visualização pública.

nota
  • O uso de um comando REORG com uma tabela de transmissão requer o Databricks Runtime 15.4 e acima.
  • Embora você possa usar a instrução REORG com qualquer tabela de transmissão, ela só é necessária ao excluir registros de uma tabela de transmissão com vetores de exclusão ativados. O comando não tem efeito quando usado com uma tabela de transmissão sem vetores de exclusão ativados.

Para excluir fisicamente os registros do armazenamento subjacente para uma tabela de transmissão com vetores de exclusão ativados, como para compliance com GDPR, os passos adicionais devem ser seguidos para garantir a execução de uma operação vacuum nos dados da tabela de transmissão.

Para excluir fisicamente registros do armazenamento subjacente:

  1. Atualize registros ou exclua registros da tabela de transmissão.
  2. Faça a execução de uma declaração REORG em relação à tabela de transmissão, especificando o parâmetro APPLY (PURGE). Por exemplo REORG TABLE <streaming-table-name> APPLY (PURGE);.
  3. Aguarde o período de retenção de dados da tabela de transmissão passar. O período default de retenção de dados é de sete dias, mas pode ser configurado com a propriedade delta.deletedFileRetentionDuration da tabela. Consulte Configurar a retenção de dados para query de viagem do tempo.
  4. REFRESH a tabela de transmissão. Consulte refresh uma tabela de transmissão. Dentro de 24 horas após a operação REFRESH, as tarefas de manutenção do pipeline, incluindo a operação VACUUM que é necessária para garantir que os registros sejam excluídos permanentemente, são de execução automática.

Monitorar execuções usando história de query​

Você pode usar a página de história de query para acessar detalhes da query e perfis de query que podem ajudá-lo a identificar queries com desempenho insatisfatório e gargalos no pipeline usado para execução das atualizações da tabela de transmissão. Para obter uma visão geral do tipo de informação disponível em histórias de query e perfis de query, consulte História de query e Perfil de query.

info

Visualização

Esse recurso está em Prévia Pública. Os administradores do Workspace podem controlar o acesso a esse recurso na página Pré-visualizações . Consulte Gerenciar prévias do Databricks.

Todas as instruções relacionadas a tabelas de transmissão aparecem na história de query. Você pode usar o filtro suspenso Instrução para selecionar qualquer comando e inspecionar as queries relacionadas. Todas as instruções CREATE são seguidas por uma instrução REFRESH que é executada de forma assíncrona em um pipeline. As instruções REFRESH normalmente incluem planos de query detalhados que fornecem percepções para otimizar o desempenho.

Para acessar as declarações REFRESH na interface do usuário da história de query, execute os passos:

  1. Clique em Ícone de história. na barra lateral esquerda para abrir a interface do usuário do Query History .
  2. Selecione a caixa de seleção REFRESH no filtro suspenso Statement .
  3. Clique no nome do comando de query para view detalhes de resumo, como a duração da query e métricas agregadas.
  4. Clique em Ver perfil da query para abrir o perfil da query. Consulte Perfil da query para obter detalhes sobre como navegar pelo perfil da query.
  5. Opcionalmente, você pode usar os links na seção Origem da query para abrir a query ou o pipeline relacionado.

Você também pode acessar detalhes de query usando links no editor de SQL ou a partir de um Notebook vinculado a um SQL Warehouse.

Acessar tabelas de transmissão de clientes externos​

Para acessar tabelas de transmissão de clientes externos do Delta Lake ou Iceberg que não suportam APIs abertas, você pode usar o Modo de Mode. O Compatibility Mode cria uma versão somente leitura da sua tabela de transmissão que pode ser acessada por qualquer cliente Delta Lake ou Iceberg.

Outros recursos​

Nesta página