Compartir a través de

CREATE STREAMING TABLE (Canalizaciones declarativas de Lakeflow)

Una tabla de streaming es una tabla compatible con el procesamiento de datos incremental o de streaming. Las tablas de streaming están respaldadas por canalizaciones declarativas de Lakeflow. Cada vez que se actualiza una tabla de streaming, los datos agregados a las tablas de origen se anexan a la tabla de streaming. Puede actualizar las tablas de streaming manualmente o según una programación.

Para más información sobre cómo realizar o programar actualizaciones, consulte Ejecutar una actualización en las canalizaciones declarativas de Lakeflow.

Sintaxis

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

    Si se especifica, creará la tabla o actualizará una tabla existente y su contenido.

  • PRIVADO

    Crea una tabla de streaming privada.

    • No se agregan al catálogo y solo son accesibles dentro de la canalización de definición
    • Pueden tener el mismo nombre que un objeto existente en el catálogo. Dentro de la canalización, si una tabla de streaming privada y un objeto del catálogo tienen el mismo nombre, las referencias al nombre se resolverán en la tabla de streaming privada.
    • Las tablas de streaming privado solo se conservan durante la duración de la canalización, no solo una sola actualización.

    Las tablas de streaming privadas se crearon anteriormente con el TEMPORARY parámetro .

  • table_name

    Nombre de la tabla recién creada. El nombre de tabla completo debe ser único.

  • especificación_de_tabla

    Esta cláusula opcional define la lista de columnas y sus tipos, propiedades, descripciones y restricciones de columnas.

  • restricción_de_tabla

    Importante

    Esta característica está en versión preliminar pública.

    Al especificar un esquema, puede definir claves principales y externas. Las restricciones son informativas y no se aplican. Consulte la cláusula CONSTRAINT en la referencia del lenguaje SQL.

    Nota:

    Para definir restricciones en las tablas, tu canalización debe estar habilitada para el Unity Catalog.

  • cláusulas_tabla

    Opcionalmente, especifique las propiedades de creación de particiones, comentarios y definidas por el usuario para la tabla. Cada subcláusula solo se puede especificar una vez.

    • USO DE DELTA

      Especifica el formato de datos. La única opción es DELTA.

      Esta cláusula es opcional y el valor predeterminado es DELTA.

    • PARTICIONADO POR

      Lista opcional de una o varias columnas que se van a usar para crear particiones en la tabla. Mutuamente excluyente con CLUSTER BY.

      La agrupación en clústeres líquidos proporciona una solución flexible y optimizada para la agrupación en clústeres. Considere usar CLUSTER BY en lugar de PARTITIONED BY para canalizaciones declarativas de Lakeflow.

    • CLUSTER BY

      Habilite la agrupación en clústeres líquidos en la tabla y defina las columnas que se usarán como claves de agrupación en clústeres. Mutuamente excluyente con PARTITIONED BY.

      Consulte Uso de clústeres líquidos para tablas.

    • UBICACIÓN

      Ubicación de almacenamiento opcional para los datos de la tabla. Si no se establece, el sistema establecerá de manera predeterminada la ubicación de almacenamiento de la canalización.

    • COMENTARIO

      Literal opcional STRING para describir la tabla.

    • TBLPROPERTIES

      Lista opcional de propiedades de la tabla.

    • CONROW FILTER

    Importante

    Esta característica está en versión preliminar pública.

    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. Esto resulta útil para el control de acceso específico, ya que permite a la función inspeccionar la identidad y las pertenencias a grupos del usuario que realiza la invocación para decidir si se filtran determinadas filas.

    Consulte ROW FILTER cláusula.

  • consulta

    Esta cláusula rellena la tabla mediante los datos de query. Esta consulta debe ser una consulta de streaming . Use la palabra clave STREAM para usar la semántica de streaming para leer desde el origen. Si la lectura encuentra un cambio o eliminación en un registro existente, se produce un error. Es más seguro leer de orígenes estáticos o de solo anexión. Para ingerir datos que tienen confirmaciones de cambios, puede usar Python y la SkipChangeCommits opción para controlar errores.

    Cuando se especifican juntos query y table_specification, el esquema de tabla especificado en table_specification debe contener todas las columnas devueltas por query, de lo contrario, se produce un error. Las columnas especificadas en table_specification pero no devueltas por query devuelven valores null cuando se consultan.

    Para más información sobre los datos de streaming, consulte Transformación de datos con canalizaciones.

Permisos necesarios

El usuario de ejecución para una canalización debe tener los permisos siguientes:

  • SELECT privilegios sobre las tablas base a las que hace referencia la tabla de streaming.
  • USE CATALOG privilegios en el catálogo primario y el privilegio USE SCHEMA en el esquema primario.
  • CREATE MATERIALIZED VIEW privilegios en el esquema de la tabla de streaming.

Para que un usuario pueda actualizar la canalización en la que se define la tabla de streaming, necesita:

  • USE CATALOG privilegios en el catálogo primario y el privilegio USE SCHEMA en el esquema primario.
  • Propiedad de la tabla de streaming o REFRESH privilegios en la tabla de streaming.
  • El propietario de la tabla de streaming debe tener el SELECT privilegio sobre las tablas base a las que hace referencia la tabla de streaming.

Para que un usuario pueda consultar la tabla de streaming resultante, necesita:

  • USE CATALOG privilegios en el catálogo primario y el privilegio USE SCHEMA en el esquema primario.
  • SELECT privilegios sobre la tabla de streaming.

Limitaciones

  • Solo los propietarios de tablas pueden actualizar las tablas de streaming para obtener los datos más recientes.
  • ALTER TABLE Los comandos no se permiten en las tablas de streaming. La definición y las propiedades de la tabla se deben modificar a través de la CREATE OR REFRESH declaración o ALTER STREAMING TABLE.
  • No se admite la evolución del esquema de tabla a través de comandos DML como INSERT INTOy MERGE .
  • Los comandos siguientes no se admiten en tablas de streaming:
    • CREATE TABLE ... CLONE <streaming_table>
    • COPY INTO
    • ANALYZE TABLE
    • RESTORE
    • TRUNCATE
    • GENERATE MANIFEST
    • [CREATE OR] REPLACE TABLE
  • No se admite el cambio de nombre de la tabla ni el cambio del propietario.
  • No se admiten columnas generadas, columnas de identidad y columnas predeterminadas.

Ejemplos

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