CREATE STREAMING TABLE (pipeline)

Uma tabela "diálogo" é uma tabela com suporte para "diálogo" ou processamento incremental de dados. As tabelas de transmissão são suportadas por pipeline. Cada vez que uma tabela de transmissão é atualizada, os dados adicionados às tabelas de origem são anexados à tabela de transmissão. Você pode refresh as tabelas de transmissão manualmente ou por meio de um programa.

Para saber mais sobre como executar uma atualização programada, consulte a execução de uma atualização pipeline.

Sintaxe

CREATE [OR REFRESH] [PRIVATE] STREAMING TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ AS query ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ column_constraint ] [, ...]
    [ , table_constraint ] [...] )

   column_properties
      { NOT NULL | COMMENT column_comment | column_constraint | MASK clause } [ ... ]

table_clauses
  { USING DELTA
    PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT view_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

Parâmetros

• REFRESH

  Se especificado, criará a tabela ou atualizará uma tabela existente e seu conteúdo.

• PRIVADO

  Cria uma tabela de transmissão privada.

  • Eles não são adicionados ao catálogo e só são acessíveis dentro do pipeline de definição
  • Eles podem ter o mesmo nome de um objeto existente no catálogo. Dentro do pipeline, se uma tabela de transmissão privada e um objeto no catálogo tiverem o mesmo nome, as referências ao nome serão resolvidas para a tabela de transmissão privada.
  • As tabelas de transmissão privadas são persistidas apenas durante a vida útil do pipeline, não apenas durante uma única atualização.

  Tabelas de transmissão privadas foram criadas anteriormente com o parâmetro TEMPORARY .

• nome_da_tabela

  O nome da tabela recém-criada. O nome da tabela totalmente qualificado deve ser exclusivo.

• especificação_de_tabela

  Esta cláusula opcional define a lista de colunas, seus tipos, propriedades, descrições e restrições de coluna.

  • identificador_de_coluna

    Os nomes das colunas devem ser exclusivos e mapeados para as colunas de saída da consulta.

  • tipo_de_coluna

    Especifica o tipo de dados da coluna. Nem todos os tipos de dados suportados pelo Databricks são suportados pelas tabelas de transmissão.

  • comentário_de_coluna

    Um literal STRING opcional que descreve a coluna. Esta opção deve ser especificada junto com column_type. Se o tipo de coluna não for especificado, o comentário da coluna será ignorado.

  • restrição_de_coluna

    Adiciona uma restrição que valida os dados conforme eles fluem para a tabela. Veja gerenciar a qualidade dos dados com expectativas pipeline.

  • Cláusula MASK

    Adiciona uma função de máscara de coluna para tornar dados confidenciais anônimos.

    Veja Filtros de linha e máscaras de coluna.

• restrição_de_tabela

  Ao especificar um esquema, você pode definir uma chave primária e uma chave estrangeira. As restrições são informativas e não são impostas. Veja a cláusula CONSTRAINT na referência da linguagem SQL.

nota

Para definir restrições de tabela, seu pipeline deve ser um pipeline habilitado para o Unity Catalog.

• cláusulas_de_tabela

  Opcionalmente, especifique particionamento, comentários e propriedades definidas pelo usuário para a tabela. Cada subcláusula só pode ser especificada uma vez.

  • USANDO DELTA

    Especifica o formato dos dados. A única opção é DELTA.

    Esta cláusula é opcional e o padrão é DELTA.

  • PARTICIONADO POR

    Uma lista opcional de uma ou mais colunas a serem usadas para particionamento na tabela. Mutuamente exclusivo com CLUSTER BY.

    clustering líquido oferece uma solução flexível e otimizada para clustering. Considere usar CLUSTER BY em vez de PARTITIONED BY para pipeline.

  • cluster POR

    Habilite clustering líquido na tabela e defina as colunas a serem usadas como chave clustering . Use clustering líquido automático com CLUSTER BY AUTO e Databricks escolhe de forma inteligente a chave clustering para otimizar o desempenho da consulta. Mutuamente exclusivo com PARTITIONED BY.

    Consulte Usar clustering líquido para tabelas.

  • Localização

    Um local de armazenamento opcional para dados da tabela. Se não estiver definido, o sistema adotará como padrão o local de armazenamento do pipeline.

  • comentário

    Um literal STRING opcional para descrever a tabela.

  • PROPRIEDADES TBL

    Uma lista opcional de propriedades de tabela para a tabela.

  • COM FILTRO DE LINHA

  Adiciona uma função de filtro de linha à tabela. Consultas futuras para essa tabela recebem um subconjunto das linhas para as quais a função é avaliada como VERDADEIRO. Isso é útil para controle de acesso refinado, porque permite que a função inspecione a identidade e as associações de grupo do usuário que faz a chamada para decidir se deve filtrar determinadas linhas.

  Veja a cláusulaROW FILTER.

• query

  Esta cláusula preenche a tabela usando os dados de query. Esta consulta deve ser uma consulta de transmissão . Use a palavra-chave transmissão para usar a semântica de transmissão para ler da fonte. Se a leitura encontrar uma alteração ou exclusão em um registro existente, um erro será gerado. A forma mais segura de ler dados é a partir de fontes estáticas ou de fontes que permitem apenas a inserção de dados. Para ingerir dados que sofreram commit de alteração, você pode adicionar a opção de leitura SkipChangeCommits para lidar com erros.

  Quando você especifica um query e um table_specification juntos, o esquema da tabela especificado em table_specification deve conter todas as colunas retornadas pelo query, caso contrário, você receberá um erro. Todas as colunas especificadas em table_specification , mas não retornadas por query , retornam valores null quando consultadas.

  Para mais informações sobre transmissão de dados, veja transformação de dados com pipeline.

  • Opções de leitura

    Você pode especificar opções de leitura na consulta para configurar como os dados são lidos da fonte. Por exemplo, você pode especificar skipChangeCommits para ignorar qualquer commit de alteração nos dados de origem. As opções de leitura são especificadas como um mapa na cláusula WITH da consulta. Por exemplo:

    SQL

    SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS=TRUE, STARTINGVERSION=X)

    O =TRUE é opcional, portanto você também pode especificar uma opção booleana assim:

    SQL

    SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS)

nota

As opções de leitura são suportadas apenas para Databricks Runtime 17.3 e versões superiores.

As opções de leitura abaixo são suportadas para Delta. Para obter detalhes sobre cada opção, consulte a tabelaDelta de leituras e gravações de transmissão.

• maxFilesPerTrigger
• maxBytesPerTrigger
• startingVersion
• startingTimestamp
• readChangeFeed
• withEventTimeOrder
• skipChangeCommits

Permissões necessárias

O usuário execução-as para um pipeline deve ter as seguintes permissões:

• SELECT privilégio sobre as tabelas base referenciadas pela tabela de transmissão.
• USE CATALOG privilégio no catálogo pai e o privilégio USE SCHEMA no esquema pai.
• CREATE MATERIALIZED VIEW privilégio no esquema da tabela de transmissão.

Para que um usuário possa atualizar o pipeline no qual a tabela de transmissão está definida, ele precisa:

• USE CATALOG privilégio no catálogo pai e o privilégio USE SCHEMA no esquema pai.
• Propriedade da tabela transmissão ou privilégio REFRESH na tabela transmissão.
• O proprietário da tabela de transmissão deve ter o privilégio SELECT sobre as tabelas base referenciadas pela tabela de transmissão.

Para que um usuário possa consultar a tabela de transmissão resultante, ele precisa:

• USE CATALOG privilégio no catálogo pai e o privilégio USE SCHEMA no esquema pai.
• SELECT privilégio sobre a tabela de transmissão.

Limitações

• Somente os proprietários das tabelas podem refresh as tabelas de transmissão para obter os dados mais recentes.

• ALTER TABLE comando não são permitidos em tabelas de transmissão. A definição e as propriedades da tabela devem ser alteradas por meio da instrução CREATE OR REFRESH ou ALTER STREAMING TABLE .

• A evolução do esquema da tabela por meio de comandos DML como INSERT INTO e MERGE não é suportada.

• Os seguintes comandos não são suportados em tabelas de transmissão:

  • CREATE TABLE ... CLONE <streaming_table>
  • COPY INTO
  • ANALYZE TABLE
  • RESTORE
  • TRUNCATE
  • GENERATE MANIFEST
  • [CREATE OR] REPLACE TABLE

• Não há suporte para renomear a tabela ou alterar o proprietário.

• Colunas geradas, colunas de identidade e colunas default não são suportadas.

Exemplos

SQL

-- Define a streaming table from a volume of files:
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- Define a streaming table from a streaming source table:
CREATE OR REFRESH STREAMING TABLE customers_silver
AS SELECT * FROM STREAM(customers_bronze)

-- Define a table with a row filter and column mask:
CREATE OR REFRESH STREAMING TABLE customers_silver (
  id int COMMENT 'This is the customer ID',
  name string,
  region string,
  ssn string MASK catalog.schema.ssn_mask_fn COMMENT 'SSN masked for privacy'
)
WITH ROW FILTER catalog.schema.us_filter_fn ON (region)
AS SELECT * FROM STREAM(customers_bronze)

-- Define a streaming table that you can add flows into:
CREATE OR REFRESH STREAMING TABLE orders;