CREATE TABLE ... FLOW (canalizaciones)

Importante

Esta característica se encuentra en su versión beta.

Use la CREATE TABLE ... FLOW instrucción para crear una tabla administrada en una canalización, escrita por uno o varios flujos.

Syntax

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 fanar varios orígenes en una tabla administrada, declare varios flujos que tienen como destino CREATE FLOW (canalizaciones)::

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

  • table_name

    Nombre de la tabla administrada que se va a crear. Si el nombre no está calificado, la tabla se crea en el esquema de destino de la canalización. El nombre aún no debe pertenecer a una tabla de streaming.

  • especificación_de_tabla

    Opcionalmente, define las columnas, sus tipos, propiedades y descripciones. Si se omite, el esquema se deduce de la consulta de flujo.

  • CONSTRAINT expectation_name EXPECT (expectation_expr) [ EN CASO DE VIOLACIÓN { FALLAR UPDATE | ELIMINAR FILA } ]

    Añade expectativas de calidad de datos a la tabla gestionada. Estas expectativas de calidad de datos pueden rastrearse a lo largo del tiempo y acceder a través del registro de eventos de la canalización. Una FAIL UPDATE expectativa hace que se produzca un error en el procesamiento al crear la tabla, así como al actualizar la tabla. Una DROP ROW expectativa hace que se quite toda la fila si no se cumple la expectativa. Consulte Administración de la calidad de los datos con las expectativas de canalización.

    expectation_expr puede estar compuesto por literales, identificadores de columna dentro de la tabla y funciones u operadores SQL deterministas incorporados excepto:

    Además, expectation_expr no deben contener ninguna subconsulta .

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

    Opcionalmente, particiona la tabla por un subconjunto de columnas.

  • CLUSTER BY cláusula

    Opcionalmente, habilita la agrupación en clústeres líquidos en la tabla. No se puede combinar PARTITIONED BY y CLUSTER BY.

  • RUTA DE UBICACIÓN

    Una ubicación de almacenamiento opcional para los datos de la tabla.

  • COMENTARIO table_comment

    Literal STRING que describe la tabla.

  • Cláusula TBLPROPERTIES

    Opcionalmente, establece una o varias propiedades de tabla definidas por el usuario.

  • Cláusula WITH ROW FILTER

    Agrega una función de filtro de fila a la tabla. Las consultas futuras de esa tabla reciben un subconjunto de las filas para las que la función se evalúa como TRUE.

  • CONSULTA FLOW INSERT [ONCE] BY NAME

    Define un flujo de anexión que inserta el resultado de query en la tabla, que coincide con las columnas de resultado en las columnas de tabla por nombre. query puede hacer referencia a orígenes por lotes o de streaming. ONCE ejecuta el flujo una sola vez (por ejemplo, para un reposición) en lugar de en cada actualización. Cada flujo con nombre procesa su entrada exactamente una vez por actualización de canalización, idéntica a FLOW INSERT BY NAME en una tabla de streaming.

  • PUNTO DE CONTROL DE IMPORTACIÓN checkpoint_path

    Un literal con la ruta hacia un punto de control de Streaming Estructurado existente para importar al flujo, de modo que un flujo migrado se reanude desde su último desplazamiento STRING comprometido en lugar de reprocesar el código fuente. Importar un punto de control está en Beta. Consulta Migrar un punto de control de Streaming Estructurado.

Migrar un punto de control de Streaming Estructurado

Importante

Importar un punto de control está en Beta.

Se utiliza IMPORT CHECKPOINT para migrar una carga de trabajo de Streaming Estructurado existente a una tubería sin reprocesar el código fuente. Configura checkpoint_path como la checkpointLocation consulta de Structured Streaming utilizada, que puede ser un almacenamiento en la nube, un volumen de Unity Catalog o una ruta DBFS. En la primera actualización de la canalización, el flujo clona ese punto de control en el almacenamiento gestionado de la canalización. El flujo reanuda entonces desde el último desplazamiento comprometido con su estado (como agregaciones, claves de deduplicación y marcas de agua) intacto. Las actualizaciones posteriores de la tubería utilizan el punto de control clonado del flujo; el punto de control original no se modifica.

Warning

Detiene la consulta original de Structured Streaming antes de ejecutar la pipeline. La consulta original de Structured Streaming puede reutilizarse tras la importación, pero necesitas gestionar su estado de punto de control y asegurarte de que la pipeline y la consulta no escriban en la misma tabla al mismo tiempo, lo que puede producir datos duplicados.

Recrea la consulta de Streaming Estructurado como un flujo de pipeline que continúa escribiendo en la tabla donde escribió la consulta, e importa su punto de control:

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

El punto de control se importa solo en la primera actualización del pipeline. Las actualizaciones posteriores ignoran IMPORT CHECKPOINT. Una actualización completa no reimporta el punto de control. Parte de un nuevo punto de control vacío y reprocesa la fuente.

Limitaciones

  • Las tablas administradas no admiten flujos de cambio CDC. AUTO CDC INTO(SQL) o apply_changes / apply_changes_from_snapshot (Python) en una tabla administrada se produce un error con MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. Para los objetivos CDC, utiliza unCREATE STREAMING TABLE (pipelines) con un flujo definido por CREATE FLOW (pipelines).
  • Las tablas administradas no admiten FLOW ... REPLACE WHERE. Solo FLOW INSERT BY NAME es compatible. Para recalcular y sobrescribir un subconjunto objetivo de una tabla, crea una tabla de flujo y define un REPLACE WHERE flujo con CREATE FLOW (pipelines). Consulte Procesamiento por lotes con flujos REPLACEWHERE.
  • Las tablas administradas solo se admiten en canalizaciones con El catálogo de Unity. No se admite el metastore de Hive.
  • No se puede reutilizar el nombre de una tabla de streaming existente para una tabla administrada. Quite primero la tabla de streaming o se produzca un error en la instrucción con CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
  • IMPORT CHECKPOINT solo se soporta en FLOW INSERT BY NAME.

Ejemplos

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

  • Última actualización el