Carregar dados usando tabelas de streaming no Databricks SQL

O Databricks recomenda usar tabelas de streaming para ingerir dados usando o Databricks SQL. Uma tabela de streaming é uma tabela registrada no Catálogo do Unity com suporte extra para streaming ou processamento de dados incremental. Um pipeline Delta Live Tables é criado automaticamente para cada tabela de streaming. Você pode usar tabelas de streaming para carregamento incremental de dados do Kafka e do armazenamento de objetos de nuvem.

Este artigo demonstra o uso de tabelas de streaming para carregar dados do armazenamento de objetos de nuvem configurados como um volume do Catálogo do Unity (recomendado) ou local externo.

Antes de começar

Antes de começar, você precisará atender aos seguintes requisitos.

Requisitos do espaço de trabalho:

• Uma conta do Azure Databricks sem servidor habilitada. Para obter mais informações, consulte Habilitar SQL warehouses sem servidor.
• Um workspace com o Catálogo do Unity habilitado. Para obter mais informações, confira Configurar e gerenciar o Catálogo do Unity.

Requisitos de computação:

Você deverá usar um dos seguintes:

• Um SQL warehouse que usa o canal Current.

• Compute com o modo de acesso compartilhado no Databricks Runtime 13.3 LTS ou superior.

• Compute com o modo de acesso de usuário único no Databricks Runtime 15.4 LTS ou superior.

  No Databricks Runtime 15.3 e inferior, você não pode usar a computação de usuário único para consultar tabelas de streaming que pertencem a outros usuários. Você pode usar a computação de usuário único no Databricks Runtime 15.3 e inferior somente se você for o proprietário da tabela de streaming. O criador da tabela é o proprietário.

  O Databricks Runtime 15.4 LTS e superior dá suporte a consultas em tabelas geradas por Delta Live Tables na computação de usuário único, independentemente da propriedade da tabela. Para aproveitar a filtragem de dados fornecida no Databricks Runtime 15.4 LTS e superior, garanta que o seu espaço de trabalho está habilitado para computação sem servidor, porque a funcionalidade de filtragem de dados que dá suporte às tabelas geradas pelo Delta Live Tables é executada na computação sem servidor. Você pode ser cobrado por recursos de computação sem servidor ao usar a computação de usuário único para executar operações de filtragem de dados. Consulte Controle de acesso refinado na computação de usuário único.

Requisitos de permissões:

• O privilégio READ FILES em um local externo do Catálogo do Unity. Para obter informações, confira Criar um local externo para conectar o armazenamento em nuvem ao Azure Databricks.
• O privilégio USE CATALOG no catálogo no qual você cria a tabela de streaming.
• O privilégio USE SCHEMA no esquema no qual você cria a tabela de streaming.
• O privilégio CREATE TABLE no esquema no qual você cria a tabela de streaming.

Outros requisitos:

• O caminho para os dados de origem.

  Exemplo de caminho de volume: /Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

  Exemplo de caminho de local externo: abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis

  Observação

  Este artigo pressupõe que os dados que você deseja carregar estão em um local de armazenamento em nuvem que corresponde a um volume do Catálogo do Unity ou ao local externo ao qual você tem acesso.

Descobrir e visualizar dados de origem

1. Na barra lateral do workspace, clique em Consultas e clique em Criar consulta.

2. No editor de consultas, selecione um SQL warehouse que usa o canal Current na lista suspensa.

3. Cole o seguinte no editor, substituindo valores em colchetes angulares (<>) pelas informações que identificam seus dados de origem e clique em Executar.

   Observação

   Você poderá encontrar erros de inferência de esquema ao executar a função read_files com valor de tabela se os padrões para a função não puderem analisar seus dados. Por exemplo, talvez seja necessário configurar o modo de várias linhas para arquivos CSV ou JSON de várias linhas. Para obter uma lista de opções de analisador, consulte Função com valor de tabela read_files.

   /* Discover your data in a volume */
   LIST "/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>"
   
   /* Preview your data in a volume */
   SELECT * FROM read_files("/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>") LIMIT 10
   
   /* Discover your data in an external location */
   LIST "abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>"
   
   /* Preview your data */
   SELECT * FROM read_files("abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>") LIMIT 10
   

Carregar dados em uma tabela de streaming

Para criar uma tabela de streaming com base em dados no armazenamento de objetos de nuvem, cole o seguinte no editor de consultas e clique em Executar:

/* Load data from a volume */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>')

/* Load data from an external location */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>')

Atualizar uma tabela de streaming usando um pipeline DLT

Esta seção descreve os padrões para atualizar uma tabela de streaming com os dados mais recentes disponíveis das fontes definidas na consulta.

Quando você CREATE ou REFRESH uma tabela de streaming, a atualização é processada usando um pipeline do Delta Live Tables sem servidor. Cada tabela de streaming definida tem um pipeline de tabelas dinâmicas do Delta associado.

Depois de executar o comando REFRESH, o link do pipeline DLT será retornado. Você pode usar o link do pipeline DLT para verificar o status da atualização.

ConsulteO que é o Delta Live Tables?.

Ingerir apenas novos dados

Por padrão, a função read_files lê todos os dados existentes no diretório de origem durante a criação da tabela e processa registros recém-chegados com cada atualização.

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

CREATE OR REFRESH STREAMING TABLE my_bronze_table
AS SELECT *
FROM STREAM read_files(
  'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
  includeExistingFiles => false)

Atualizar totalmente uma tabela de streaming

As atualizações completas reprocessam todos os dados disponíveis na origem com a definição mais recente. Não é recomendável chamar atualizações completas em origens que não mantêm todo o histórico dos dados ou têm períodos de retenção curtos, como Kafka, pois 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 origem.

Por exemplo:

REFRESH STREAMING TABLE my_bronze_table FULL

Agendar uma tabela de streaming para atualização automática

Para configurar uma tabela de streaming para atualizar automaticamente com base em um agendamento definido, cole o seguinte no editor de consultas e clique em Executar:

ALTER STREAMING TABLE
[[<catalog>.]<database>.]<name>
ADD [SCHEDULE [REFRESH]
        CRON '<cron-string>'
                [ AT TIME ZONE '<timezone-id>' ]];

Para obter exemplos de consultas de agendamento de atualização, consulte ALTER STREAMING TABLE.

Acompanhar o status de uma atualização

Você pode exibir o status de uma atualização de tabela de streaming visualizando o pipeline que gerencia a tabela de streaming na interface do usuário do Delta Live Tables ou exibindo as Informações de Atualização retornadas pelo comando DESCRIBE EXTENDED da tabela de streaming.

DESCRIBE EXTENDED <table-name>

Ingestão de streaming do Kafka

Para obter um exemplo de ingestão de streaming do Kafka, consulte read_kafka.

Conceder aos usuários acesso a uma tabela de streaming

Para conceder aos usuários o privilégio SELECT na tabela de streaming para que eles possam consultá-la, cole o seguinte no editor de consultas e clique em Executar:

GRANT SELECT ON TABLE <catalog>.<schema>.<table> TO <user-or-group>

Para obter mais informações sobre concessão de privilégios no Catálogo do Unity, confira Privilégios e objetos protegíveis do Catálogo do Unity.

Recursos adicionais

• Tabela de streaming
• Função read_files com valor de tabela
• CREATE STREAMING TABLE
• ALTER STREAMING TABLE
• Função com valor de tabela read_kafka