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

CREATE TABLE ... FLOW (pipelines)

info

Beta

Este recurso está em Beta.

Use a instrução CREATE TABLE ... FLOW para criar uma tabela gerenciada em um pipeline, escrita por um ou mais fluxos.

Sintaxe​

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

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

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

Para direcionar várias fontes para uma única tabela gerenciada, declare vários fluxos que a segmentam com CREATE FLOW (pipelines):

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parâmetros​

  • table_name

    O nome da tabela gerenciada a ser criada. Se o nome não for qualificado, a tabela será criada no esquema de destino do pipeline. O nome não deve pertencer a uma tabela de transmissão.

  • table_specification

    Opcionalmente, define as colunas, seus tipos, propriedades e descrições. Se omitido, o esquema é inferido da query do fluxo.

  • CONSTRAINT expectation_name EXPECT (expectation_expr) [ ON VIOLATION { FAIL UPDATE | DROP ROW } ]

    Adiciona expectativas de qualidade de dados à tabela gerenciada. Essas expectativas de qualidade de dados podem ser acompanhadas ao longo do tempo e acessadas por meio do log de eventos do pipeline. Uma expectativa FAIL UPDATE faz com que o processamento falhe tanto na criação quanto na atualização da tabela. Uma expectativa DROP ROW faz com que a linha inteira seja descartada se a expectativa não for atendida. Consulte Gerenciar a qualidade dos dados com expectativas de pipeline.

    expectation_expr pode ser composto por literais, identificadores de coluna dentro da tabela e funções ou operadores SQL integrados e determinísticos, exceto:

    Além disso, expectation_expr não deve conter nenhuma subconsulta.

  • PARTICIONADO POR (col [, ...])

    Opcionalmente, particiona a tabela por um subconjunto de colunas.

  • Cláusula CLUSTER BY

    Opcionalmente ativa clustering líquido na tabela. Não é possível combinar PARTITIONED BY e CLUSTER BY.

  • Caminho do local

    Um local de armazenamento opcional para os dados da tabela.

  • COMMENT table_comment

    Um literal STRING que descreve a tabela.

  • Cláusula TBLPROPERTIES

    Define opcionalmente uma ou mais propriedades de tabela definidas pelo usuário.

  • Cláusula WITH ROW FILTER

    Adiciona uma função de filtro de linha à tabela. Futuras queries para essa tabela recebem um subconjunto das linhas para as quais a função avalia como TRUE.

  • FLOW INSERT [ONCE] BY NAME query

    Define um fluxo de acréscimo que insere o resultado de query na tabela, correspondendo colunas de resultado a colunas da tabela pelo nome . query pode fazer referência a fontes de lotes ou de transmissão. ONCE executa o fluxo uma única vez (por exemplo, para um preenchimento retroativo) em vez de a cada atualização. Cada fluxo nomeado processa sua entrada exatamente uma vez por atualização de pipeline, idêntico a FLOW INSERT BY NAME em uma tabela de transmissão.

  • IMPORT CHECKPOINT checkpoint_path

    Um literal STRING com o caminho para um ponto de verificação de Structured Streaming existente para importar para o fluxo, para que uma transmissão migrada retome a partir do último deslocamento confirmado em vez de reprocessar a fonte. A importação de um ponto de verificação está em Beta. Consulte Migrar um ponto de verificação de Structured Streaming.

Migrar um ponto de verificação de Structured Streaming​

info

Beta

A importação de um ponto de verificação está em Beta.

Use IMPORT CHECKPOINT para migrar uma carga de trabalho de Structured Streaming existente para um pipeline sem processar novamente a fonte. Defina checkpoint_path como checkpointLocation usada pela sua query de Structured Streaming, que pode ser um armazenamento na cloud, um volume do Unity Catalog ou um caminho DBFS. Na primeira atualização do pipeline, o fluxo clona esse ponto de verificação no armazenamento gerenciado pelo pipeline. O fluxo então retoma do último deslocamento confirmado com seu estado (como agregações, keys de deduplicação e marcas d'água) intacto. As atualizações subsequentes do pipeline usam o ponto de verificação clonado do fluxo; o ponto de verificação original não é modificado.

atenção

Pare a query de Structured Streaming original antes de executar o pipeline. A query de Structured Streaming original pode ser reutilizada após a importação, mas você precisa gerenciar seu estado de ponto de verificação e garantir que o pipeline e a query não escrevam na mesma tabela ao mesmo tempo, o que pode produzir dados duplicados.

Recrie a query de Structured Streaming como um fluxo de pipeline que continua escrevendo na tabela na qual a query escreveu e importe seu ponto de verificação:

SQL
-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

O ponto de verificação é importado apenas na primeira atualização do pipeline. Atualizações posteriores ignoram IMPORT CHECKPOINT. Uma refresh completa não reimporta o ponto de verificação. Começa a partir de um novo ponto de verificação vazio e reprocessa a fonte.

Limitações​

  • Tabelas gerenciadas do not support CDC change flows. AUTO CDC INTO (SQL) or apply_changes / apply_changes_from_snapshot (Python) against a tabela gerenciada fails with MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. For CDC targets, use a CREATE STREAMING TABLE (pipelines) with a flow defined by CREATE FLOW (pipelines).
  • As tabelas gerenciadas não oferecem suporte a FLOW ... REPLACE WHERE. Apenas FLOW INSERT BY NAME é suportado. Para recalcular e substituir um subconjunto segmentado de uma tabela, crie uma tabela de transmissão e defina um fluxo REPLACE WHERE com CREATE FLOW (pipelines). Consulte Processamento em lotes com fluxos REPLACE WHERE.
  • Tabelas gerenciadas são compatíveis apenas em pipelines com o Unity Catalog. O Hive metastore não é compatível.
  • Não é possível reutilizar o nome de uma tabela de transmissão existente para uma tabela gerenciada. Descarte a tabela de transmissão primeiro, ou a declaração falhará com CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
  • IMPORT CHECKPOINT é suportado apenas em FLOW INSERT BY NAME.

Exemplos​

SQL
-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

Recursos adicionais​

Nesta página