CREATE TABLE ... FLOW (oleodutos)

Importante

Este recurso está em versão Beta.

Use a CREATE TABLE ... FLOW instrução para criar uma tabela gerida num 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 ventilar múltiplas fontes numa tabela gerida, declare vários fluxos que a direcionam com CREATE FLOW (pipelines):

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    O nome da tabela gerida a criar. Se o nome não for qualificado, a tabela é criada no esquema alvo do pipeline. O nome não deve já pertencer a uma tabela de streaming.

  • especificação_da_tabela

    Opcionalmente define as colunas, os seus tipos, propriedades e descrições. Se omitido, o esquema é inferido a partir da consulta de fluxo.

  • CONSTRAINT expectation_name EXIGIR (expectation_expr) [ EM CASO DE VIOLAÇÃO { FAIL UPDATE | DROP ROW } ]

    Adiciona expectativas de qualidade dos dados à tabela gerida. Estas expectativas de qualidade dos dados podem ser acompanhadas ao longo do tempo e acedidas através do registo de eventos do pipeline. Uma expectativa de FAIL UPDATE faz com que o processamento falhe ao criar a tabela e ao atualizá-la. Uma DROP ROW expectativa faz com que toda a linha seja removida se a expectativa não for cumprida. Consulte Gerir a qualidade dos dados com as expectativas do fluxo de dados.

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

    Também expectation_expr não deve conter nenhuma subconsulta.

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

    Opcionalmente, particiona a tabela por um subconjunto de colunas.

  • CLUSTER BY cláusula

    Opcionalmente, ativa o agrupamento de líquidos na mesa. Não se pode combinar PARTITIONED BY e CLUSTER BY.

  • Caminho de LOCALIZAÇÃO

    Um local opcional de armazenamento para os dados da tabela.

  • COMENTÁRIO table_comment

    Uma STRING descrição literal da mesa.

  • Cláusula TBLPROPERTIES

    Opcionalmente, define uma ou mais propriedades de tabela definidas pelo utilizador.

  • cláusula WITH ROW FILTER

    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 avalia para TRUE.

  • CONSULTA DE FLUXO INSERT [ONCE] POR NOME

    Define um fluxo de adição que insere o resultado de query na tabela, associando as colunas de resultado às colunas da tabela pelo nome. query pode consultar fontes em lote ou em streaming. ONCE Executa o fluxo uma única vez (por exemplo, para um preenchimento) em vez de em todas as atualizações. Cada fluxo nomeado processa a sua entrada exatamente uma vez por atualização de pipeline, idêntico a FLOW INSERT BY NAME uma tabela de streaming.

  • PONTO DE CONTROLO DE IMPORTAÇÃO checkpoint_path

    Um STRING literal com o caminho para um checkpoint de Streaming Estruturado existente para importar no fluxo, de modo a que um fluxo migrado retome a partir do seu último offset comprometido em vez de reprocessar a fonte. Importar um checkpoint está em Beta. Veja Migrar um ponto de verificação de Streaming Estruturado.

Migre um ponto de verificação de Streaming Estruturado

Importante

Importar um checkpoint está em Beta.

Use IMPORT CHECKPOINT para migrar uma carga de trabalho de Streaming Estruturado existente para um pipeline sem reprocessar a fonte. Defina checkpoint_path para a checkpointLocation sua consulta de Streaming Estruturado utilizada, que pode ser um armazenamento na cloud, um volume do Unity Catalog ou um caminho DBFS. Na primeira atualização do pipeline, o flow clona esse checkpoint no armazenamento gerido do pipeline. O fluxo retoma então a partir do último deslocamento comprometido com o seu estado (como agregações, chaves de deduplicação e marcas de água) intacto. As atualizações subsequentes do pipeline utilizam o checkpoint clonado do fluxo; o checkpoint original não é modificado.

Warning

Pare a consulta original de Structured Streaming antes de executar o pipeline. A consulta original de Structured Streaming pode ser reutilizada após a importação, mas é necessário gerir o estado do checkpoint e garantir que o pipeline e a consulta não escrevem na mesma tabela ao mesmo tempo, o que pode produzir dados duplicados.

Recrie a consulta de Streaming Estruturado como um fluxo de pipeline que continua a escrever na tabela onde a consulta escreveu, e importa o seu checkpoint:

-- 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 checkpoint é importado apenas na primeira atualização do pipeline. Atualizações posteriores ignoram IMPORT CHECKPOINTo . Uma atualização completa não reimporta o checkpoint. Começa a partir de um novo ponto de controlo vazio e reprocessa a fonte.

Limitações

  • As tabelas geridas não suportam fluxos de alteração CDC. AUTO CDC INTO(SQL) ou apply_changes / apply_changes_from_snapshot (Python) contra uma tabela gerida falha com .MANAGED_TABLE_DOES_NOT_SUPPORT_CDC Para alvos CDC, use um CREATE STREAMING TABLE (pipelines) com um fluxo definido por CREATE FLOW (pipelines).
  • As tabelas geridas não suportam FLOW ... REPLACE WHERE. Só o FLOW INSERT BY NAME é suportado. Para recalcular e sobrescrever um subconjunto alvo de uma tabela, crie-se uma tabela de streaming e defina um REPLACE WHERE fluxo com CREATE FLOW (pipelines). Consulte o processamento em lote com fluxos REPLACE WHERE.
  • As tabelas geridas são suportadas apenas em pipelines com o Unity Catalog. A metastore Hive não é suportada.
  • Não pode reutilizar o nome de uma tabela de streaming existente para uma tabela gerida. Elimine primeiro a tabela de streaming, ou a instrução falha com CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
  • IMPORT CHECKPOINT é suportado apenas em FLOW INSERT BY NAME.

Exemplos

-- 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

  • Última atualização a