Utilize tabelas autónomas de transmissão

Uma tabela de streaming autónoma é uma tabela registada no Unity Catalog com suporte extra para streaming ou processamento incremental de dados, definida fora de um pipeline Lakeflow. Um pipeline é criado automaticamente para cada tabela de streaming. Você pode usar tabelas de streaming para carregamento incremental de dados do Kafka e armazenamento de objetos na nuvem.

Pode criar e atualizar tabelas de streaming independentes a partir de um armazém SQL do Databricks ou de um notebook executado em computação geral sem servidor. Para detalhes sobre as diferenças entre as duas opções de computação, veja Requisitos para pipelines independentes.

Para criar e atualizar tabelas de streaming autónomas com Python a partir de um caderno, veja Usar Python com pipelines independentes.

Requerimentos

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

Criar tabelas de streaming

Uma tabela de streaming é definida por uma consulta SQL no Databricks SQL. Quando você cria uma tabela de streaming, os dados atualmente nas tabelas de origem são usados para criar a tabela de streaming. Depois disso, você atualiza a tabela, geralmente em um cronograma, para obter quaisquer dados adicionados nas tabelas de origem para anexar à tabela de streaming.

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

Para criar uma tabela de streaming a partir de uma tabela existente, use a CREATE STREAMING TABLE instrução, como no exemplo a seguir:

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

Nesse caso, a tabela sales de streaming é criada a partir de colunas específicas da tabela raw_data, com um agendamento para ser atualizada a cada hora. A consulta usada deve ser uma consulta de streaming . Use a palavra-chave STREAM para aplicar as semânticas de streaming na leitura a partir da fonte.

Computação usada para atualização

Quando você cria uma tabela de streaming usando a CREATE OR REFRESH STREAMING TABLE instrução, a atualização inicial de dados e a população começam imediatamente. Estas operações não consomem o cálculo do SQL warehouse do Databricks. Em vez disso, as tabelas de streaming dependem de pipelines sem servidor tanto para criação como para atualização. Um pipeline dedicado sem servidor é criado e gerenciado automaticamente pelo sistema para cada tabela de streaming.

Carregar ficheiros com o Auto Loader

Para criar uma tabela de streaming a partir de arquivos em um volume, use o Auto Loader. Use o Auto Loader para a maioria das tarefas de ingestão de dados do armazenamento de objetos na nuvem. O Auto Loader e os pipelines são projetados para carregar dados cada vez maiores de forma incremental e idimpotente à medida que chegam ao armazenamento em nuvem.

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

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:

CREATE OR REFRESH STREAMING TABLE sales
  SCHEDULE EVERY 1 hour
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/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 no SQL, com exemplos, consulte Carregar dados do armazenamento de objetos.

Ingestão de streaming de outras fontes

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

Aplicar captura de dados de alteração (CDC) com fluxos de CDC automáticos

Utilize uma cláusula FLOW AUTO CDC para processar registos de captura de dados de alteração (CDC) de uma fonte para uma tabela de streaming. Anteriormente, a MERGE INTO instrução era comumente usada para processar registros CDC no Azure Databricks. No entanto, MERGE INTO pode produzir resultados incorretos devido a registros fora de sequência ou requer lógica complexa para reordenar registros. Ver Alterar captura de dados e instantâneos.

AUTO CDC simplifica o CDC ao tratar automaticamente os registos fora de ordem. Especifica chaves para identificar registos, uma coluna de sequência para ordenação e se deve armazenar os resultados como SCD tipo 1 (atualizações diretas) ou SCD tipo 2 (rastreio de histórico).

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

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 seguinte utiliza o SCD tipo 2 para manter um histórico de alterações:

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 detalhes completos sobre as opções e comportamentos do Auto CDC, consulte As APIs do AUTO CDC: Simplificar a captura de dados de alterações com pipelines. Para a referência sintática completa, veja CREATE STREAMING TABLE.

Aplicar substituição seletiva em lote com fluxos REPLACE WHERE

Use a FLOW REPLACE WHERE cláusula para recalcular e sobrescrever um subconjunto alvo de uma tabela de streaming sem reprocessar todo o histórico da sua tabela. REPLACE WHERE Os fluxos são bem adequados para processamento em lote incremental de junções e agregações, dados de chegada tardia, reprocessamento a montante, evolução de esquemas e preenchimentos.

Para obter informações completas sobre os fluxos REPLACE WHERE, incluindo requisitos, substituições de predicados e atualização incremental, consulte REPLACE WHERE fluxos para tabelas de streaming autónomas.

Aplicar substituição parcial de instantâneo utilizando fluxos REPLACE USING

Use a FLOW REPLACE USING cláusula para manter uma tabela de streaming sincronizada com um fluxo de snapshots parciais. Em cada atualização, um fluxo REPLACE USING substitui todas as linhas que correspondem com base nas colunas‑chave especificadas e mantém inalteradas todas as restantes linhas. Uma SEQUENCE BY coluna ordena as atualizações para que a sequência mais alta de uma chave prevaleça sempre, mesmo quando as atualizações chegam fora de ordem. Por exemplo:

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. Faz a correspondência das colunas com base no nome e não pela posição.

REPLACE USING comporta-se da mesma forma para tabelas de streaming autónomas, tal como nos pipelines do Lakeflow. Para obter informações sobre o funcionamento, a sequência, as expectativas, as limitações e exemplos, consulte Substituição parcial de instantâneos com fluxos REPLACE USING. As seguintes diferenças aplicam-se a tabelas de streaming autónomas:

• Defina o fluxo em SQL. Crie o fluxo REPLACE USING com a cláusula SQL inline FLOW REPLACE USING em CREATE OR REFRESH STREAMING TABLE. A instrução CREATE FLOW autónoma é um constructo de um pipeline do Lakeflow e não é utilizada em tabelas de streaming autónomas.
• A computação é gerida automaticamente. Tabelas autónomas de streaming são executadas em pipelines sem servidor geridos pelo sistema e requerem o Databricks Runtime 18.2 ou superior. Não escolhes entre computação clássica e serverless.

Ingerir apenas novos dados

Por defeito, a read_files função lê todos os dados existentes na pasta de origem durante a criação da tabela e depois processa os registos recém-chegados a cada atualização.

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

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 streaming são sempre executadas na versão mais recente do runtime do Databricks SQL. A propriedade pipelines.channel da tabela, anteriormente utilizada para selecionar um canal de runtime preview ou current, já não é suportada e não tem efeito. Se uma definição existente incluir esta propriedade, ela é ignorada em segurança e não é necessário removê-la.

Ocultar dados confidenciais

Você pode usar tabelas de streaming para ocultar dados confidenciais dos usuários que acessam a tabela. Uma abordagem é definir a consulta para que ela exclua colunas ou linhas confidenciais por completo. Como alternativa, você pode aplicar máscaras de coluna ou filtros de linha com base nas permissões do usuário que está consultando. Por exemplo, pode esconder a tax_id coluna para utilizadores que não estão no grupo HumanResourcesDept. Para fazer isto, use a sintaxe ROW FILTER e MASK durante a criação da tabela de streaming. Para obter mais informações, consulte Filtros de linha e máscaras de coluna.

Atualizar uma tabela de streaming

As tabelas de streaming criam e utilizam automaticamente pipelines serverless para processar operações de atualização. A atualização é gerida pelo pipeline e a atualização é monitorizada pelo armazém SQL Databricks usado para criar a tabela de streaming. As tabelas de streaming podem ser atualizadas usando um pipeline que é executado conforme uma agenda.

Mesmo que você tenha uma atualização agendada, pode chamar uma atualização manual a qualquer momento. As atualizações são tratadas pelo mesmo pipeline que foi criado automaticamente junto com a tabela de streaming.

Para atualizar uma tabela de streaming:

REFRESH STREAMING TABLE sales;

Você pode verificar o status da atualização mais recente com DESCRIBE TABLE EXTENDED.

Para saber como agendar uma atualização, consulte Atualizações de Agendamento. As atualizações agendadas podem ter notificações de atualização, e pode definir o modo de desempenho para a atualização.

Como funciona a atualização

Uma atualização de tabela em streaming só avalia as novas linhas que chegaram após a última atualização e acrescenta apenas os novos dados.

Cada atualização usa a definição atual da tabela de streaming para processar esses novos dados. Modificar uma definição de tabela de streaming não recalcula automaticamente os dados existentes. Se uma modificação for incompatível com dados existentes (por exemplo, alterar um tipo de dado), a atualização seguinte falha com um erro.

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

• Remover um filtro não reprocessa as linhas previamente filtradas.
• Alterar as projeções das colunas não afeta a forma como os dados existentes foram processados.
• As uniões com instantâneos estáticos usam o estado do instantâneo no momento do processamento inicial. Os dados que chegam tarde e que coincidiriam com o snapshot atualizado são ignorados. Isto pode levar a que os factos sejam perdidos se as dimensões estiverem atrasadas.
• Modificar o CAST de uma coluna existente resulta num erro.

Se os dados forem alterados de uma forma que não possa ser suportada na tabela de streaming existente, você poderá executar uma atualização completa.

Atualizar totalmente uma tabela de streaming

As atualizações completas reprocessam todos os dados disponíveis na fonte com a definição mais recente. Não é recomendável chamar atualizações completas em fontes que não mantêm todo o histórico dos dados ou têm períodos de retenção curtos, como Kafka, porque a atualização completa trunca os dados existentes. Talvez não seja possível recuperar dados antigos se os dados não estiverem mais disponíveis na fonte.

Por exemplo:

REFRESH STREAMING TABLE sales FULL;

Agendar e monitorizar atualizações

Pode atualizar automaticamente uma tabela de streaming conforme um horário ou quando os dados a montante mudam, e pode configurar tempos de atualização, notificações e modos de desempenho. Consulte Atualizações de horários.

Controlar o acesso a tabelas de streaming

As tabelas de streaming suportam controles de acesso avançados para dar suporte ao compartilhamento de dados, evitando a exposição de dados potencialmente privados. Um proprietário de mesa de streaming ou um usuário com o MANAGE privilégio pode conceder SELECT privilégios a outros usuários. Os utilizadores tendo SELECT acesso à tabela de streaming não precisam SELECT de acesso às tabelas referenciadas pela tabela de streaming. Esse controle de acesso permite o compartilhamento de dados enquanto controla o acesso aos dados subjacentes.

Também pode alterar o proprietário de uma tabela de streaming.

Conceder privilégios a uma tabela de streaming

Para conceder acesso a uma tabela de streaming, utilize a GRANT instrução:

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

O privilege_type pode ser:

• SELECT - o utilizador pode SELECT a tabela de streaming.
• REFRESH - o utilizador pode REFRESH a tabela de streaming. As atualizações são executadas usando as permissões do proprietário.

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

CREATE OR REFRESH STREAMING TABLE st_name AS SELECT * FROM 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 a concessão de privilégios em objetos securáveis do Unity Catalog, consulte a referência de privilégios do Unity Catalog.

Revogar privilégios numa tabela de streaming

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

REVOKE privilege_type ON <st_name> FROM principal;

Quando SELECT os privilégios em uma tabela de origem são revogados do proprietário da tabela de streaming ou de qualquer outro usuário que tenha recebido MANAGE ou SELECT privilégios na tabela de streaming, ou a tabela de origem é eliminada, o proprietário da tabela de streaming ou o usuário com acesso concedido ainda possa consultar a tabela de streaming. No entanto, ocorre o seguinte comportamento:

• O proprietário da mesa de streaming ou outros que perderam o acesso a uma mesa de streaming já não REFRESH conseguem essa mesa de streaming, e a mesa de streaming torna-se obsoleta com o tempo.
• Se automatizado com uma agenda, a próxima tarefa agendada REFRESH falha ou não é executada.

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

REVOKE SELECT ON st_name FROM read_only_user;

Alterar o proprietário de uma tabela de streaming

Um utilizador com MANAGE permissões numa tabela de streaming autónoma pode definir um novo proprietário através do Explorador de Catálogos. O novo proprietário pode ser ele próprio ou um principal de serviço no qual tem o papel de Utilizador Principal de Serviço .

1. A partir do seu espaço de trabalho Azure Databricks, clique Catalogar para abrir o Explorador de Catálogos.

2. Selecione a tabela de streaming que quer atualizar.

3. Na barra lateral direita, em Sobre esta tabela de streaming, encontre o Proprietário e clique editar.

   Observação

   Se receber uma mensagem a indicar que deve alterar o proprietário mudando o utilizador Executar como nas definições da pipeline, isso significa que a tabela de streaming está definida numa pipeline Lakeflow, e não como uma tabela autónoma. A mensagem inclui um link para as configurações do pipeline, onde é possível alterar o utilizador de execução.

4. Selecione um novo proprietário para a tabela de streaming.

   Os proprietários têm privilégios automaticamente de MANAGE e SELECT nas tabelas de streaming que possuem. Se definir um principal de serviço como proprietário de uma tabela de streaming cuja propriedade já é sua, e não tiver explicitamente SELECT ou MANAGE privilégios na tabela de streaming, então esta alteração causará a perda de todo o acesso à tabela de streaming. Neste caso, é solicitado que conceda explicitamente esses privilégios.

   Selecione tanto Conceder GERIR como Conceder SELECT privilégios para aplicar isso no Guardar.

5. Clica em Guardar para mudar o proprietário.

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

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

Se alterar o proprietário e o novo proprietário não tiver acesso às tabelas fonte ou tiver os privilégios revogados nas tabelas de origem subjacentes, os utilizadores ainda poderão consultar a tabela de streaming. No entanto:

• Não podem REFRESH usar a mesa de streaming.
• A próxima atualização agendada da tabela de streaming falha.

Perder o acesso aos dados de origem impede as atualizações, mas não impede que a tabela de streaming existente seja lida.

Excluir permanentemente registros de uma tabela de streaming

Para excluir fisicamente os registros do armazenamento subjacente de uma tabela de streaming com vetores de exclusão habilitados, como para conformidade com o GDPR, etapas adicionais devem ser tomadas para garantir que uma VACUUM operação seja executada nos dados da tabela de streaming.

Para excluir fisicamente os registros do armazenamento subjacente:

1. Atualize registros ou exclua registros da tabela de streaming.
2. Execute uma instrução REORG na tabela de streaming, especificando o parâmetro APPLY (PURGE). Por exemplo, REORG TABLE <streaming-table-name> APPLY (PURGE);.
3. Aguarde até que o período de retenção de dados da tabela de streaming passe. O período de retenção de dados padrão é de sete dias, mas pode ser configurado com a propriedade delta.deletedFileRetentionDuration table. Consulte Configuração de retenção de dados para consultas de viagem no tempo.
4. REFRESH a tabela de streaming. Consulte Atualizar uma tabela de streaming. Dentro de 24 horas após a operação, as tarefas de manutenção do gasoduto, incluindo a REFRESH operação necessária para garantir que os VACUUM registros sejam excluídos permanentemente, são executadas automaticamente.

Monitoriza a execução usando o histórico de consultas

Você pode usar a página de histórico de consultas para acessar detalhes de consulta e perfis de consulta que podem ajudá-lo a identificar consultas de baixo desempenho e gargalos no pipeline usado para executar as atualizações da tabela de streaming. Para obter uma visão geral do tipo de informações disponíveis em históricos de consultas e perfis de consulta, consulte Histórico de consultas e Perfil de consulta.

Todas as instruções relacionadas a tabelas de streaming aparecem no histórico de consultas. Você pode usar o Instrução filtro suspenso para selecionar qualquer comando e inspecionar as consultas relacionadas. Todas as CREATE declarações são seguidas por uma REFRESH declaração que é executada de forma assíncrona num pipeline. As REFRESH instruções geralmente incluem planos de consulta detalhados que fornecem informações sobre a otimização do desempenho.

Para aceder a declarações REFRESH na interface do utilizador do histórico de consultas, siga os seguintes passos:

1. Clique no na barra lateral esquerda para abrir a interface do usuário do Histórico de Consultas .
2. Selecione a caixa de seleção REFRESH no filtro de lista suspensa Statement.
3. Clique no nome da instrução de consulta para exibir detalhes de resumo, como a duração da consulta e métricas agregadas.
4. Clique em Ver perfil de consulta para abrir o perfil de consulta. Consulte Perfil de consulta para obter detalhes sobre como navegar no perfil de consulta.
5. Opcionalmente, pode usar os links na secção Origem da Consulta para abrir a consulta ou o pipeline relacionado.

Você também pode acessar os detalhes da consulta usando links no editor SQL ou a partir de um bloco de anotações anexado a um SQL warehouse.

Aceder a tabelas de transmissão a partir de clientes externos

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

Recursos adicionais

• Pipelines Declarativos Spark
• read_files função com valor de tabela
• read_kafka função com valor de tabela
• CREATE STREAMING TABLE
• ALTER STREAMING TABLE
• Use vistas materializadas independentes
• AUTO CDC APIs: Simplifique a captura de dados de mudanças através de pipelines