CREATE STREAMING TABLE (canalizaciones)

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. 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 de canalización.

Syntax

CREATE [OR REFRESH] [PRIVATE] STREAMING TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ {flow_clause | AS query} ]

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

   column_properties
      { NOT NULL | GENERATED ALWAYS AS ( expr ) | GENERATED { ALWAYS | BY DEFAULT } AS IDENTITY [ ( [ START WITH start | INCREMENT BY step ] [ ...] ) ] | DEFAULT default_expression | 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 } } [ ... ]
   } [ ... ]

flow_clause
  FLOW { { INSERT [ONCE] BY NAME query } |
  { AUTO CDC auto_cdc_flow_spec } |
  { REPLACE WHERE predicate BY NAME query } }

Parámetros

REFRESH

Si se especifica, crea la tabla o actualiza una tabla existente y su contenido.
PRIVADO

Crea una tabla de streaming privada.
- No se añaden al catálogo y solo son accesibles dentro del flujo de trabajo 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 resuelven 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.
- column_identifier
  
  Los nombres de columna deben ser únicos y asignarse a las columnas de salida de la consulta.
- tipo_de_columna
  
  Especifica el tipo de datos de la columna. No todos los tipos de datos admitidos por Azure Databricks son compatibles con las tablas de streaming.
- column_comment
  
  Un literal de STRING opcional que describe la columna. Esta opción debe especificarse junto con column_type. Si no se especifica el tipo de columna, se omite el comentario de la columna.
- GENERADO SIEMPRE COMO ( expr )
  
  Cuando se especifica esta cláusula, el valor de esta columna viene determinado por el parámetro expr especificado.
  
  El DEFAULT COLLATION de la tabla debe ser UTF8_BINARY.
  
  expr puede estar compuesto por literales, identificadores de columna dentro de la tabla y funciones u operadores SQL integrados y deterministas, a excepción de lo siguiente:
  - Funciones de agregado
  - Funciones de ventana analítica
  - Funciones de ventana de clasificación
  - Funciones generadoras de valores de tabla
  - Columnas con una intercalación distinta de UTF8_BINARY
  Además, expr no deben contener ninguna subconsulta .
- GENERATED { ALWAYS | DE FORMA PREDETERMINADA } AS IDENTITY [ ( [ START WITH start ] [ INCREMENT BY step ] ) ] ]
  
  Se aplica a: Databricks SQL Databricks Runtime 10.4 LTS y versiones posteriores
  
  Define una columna de identidad. Si escribe en la tabla y no proporciona valores para la columna de identidad, se le asignará automáticamente un valor único y estadísticamente creciente (o decreciente si step es negativo). Esta cláusula solo se admite para las tablas Delta. Además, solo se puede usar para columnas con el tipo de datos BIGINT.
  
  Los valores asignados automáticamente comienzan por start y se incrementan en step. Los valores asignados son únicos, pero no se garantiza que sean contiguos. Ambos parámetros son opcionales, con un valor predeterminado de 1. step no puede ser 0.
  
  Si los valores asignados automáticamente están fuera del intervalo del tipo de columna de identidad, se producirá un error en la consulta.
  
  Cuando se usa ALWAYS, no puede proporcionar sus propios valores para la columna de identidad.
  
  No se admiten estas operaciones:
  - PARTITIONED BY una columna de identidad
  - UPDATE una columna de identidad
  Nota:
  
  Declarar una columna de identidad en una tabla deshabilita las transacciones simultáneas. Utilice columnas de identidad únicamente en casos en los que no se requieran escrituras simultáneas en la tabla de destino.
- DEFAULT expresión_predeterminada
  
  Se aplica a: Databricks SQL Databricks Runtime 11.3 LTS y versiones posteriores
  
  Define un valor DEFAULT para la columna que se usa en INSERT, UPDATE y MERGE ... INSERT cuando no se especifica la columna.
  
  Si no se especifica ningún valor predeterminado, se aplicará DEFAULT NULL a las columnas que admiten un valor NULL.
  
  default_expression puede estar compuesto de literales y funciones u operadores de SQL integrados, excepto:
  - Funciones de agregado
  - Funciones de ventana analítica
  - Funciones de ventana de clasificación
  - Funciones generadoras de valores de tabla
  Además, default_expression no deben contener ninguna subconsulta .
  
  DEFAULT es compatible con orígenes CSV, JSON, PARQUET y ORC.
- column_constraint
  
  Agrega una clave principal informativa o una restricción de clave externa informativa a la columna de una tabla de streaming.
- Cláusula MASK
  
  Agrega una función de máscara de columna para anonimizar datos confidenciales.
  
  Consulte Filtros de fila y máscaras de columna.
- CONSTRAINT expectation_name EXPECT (expectation_expr) [ EN CASO DE VIOLACIÓN { FALLAR UPDATE | ELIMINAR FILA } ]
  
  Agrega expectativas de calidad de datos a la tabla de streaming. Se puede realizar un seguimiento de estas expectativas de calidad de los datos se pueden realizar a lo largo del tiempo y acceder a ellas a través del registro de eventos de la tabla de streaming. 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 integrados y deterministas, a excepción de lo siguiente:
  - Funciones de agregado
    - Funciones de ventana analítica
    - Funciones de ventana de clasificación
    - Funciones generadoras de valores de tabla
  Además, expr no deben contener ninguna subconsulta .
restricción_de_tabla

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 la posibilidad de usar CLUSTER BY en lugar de PARTITIONED BY para canalizaciones.
- 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. Use la agrupación automática en clústeres líquidos con CLUSTER BY AUTOy Databricks elige de forma inteligente las claves de agrupación en clústeres para optimizar el rendimiento de las consultas. 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 tiene como valor predeterminado la ubicación de almacenamiento de canalización.
- COMENTARIO
  
  Literal opcional STRING para describir la tabla.
- TBLPROPERTIES
  
  Lista opcional de propiedades de la tabla.
- CONROW 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. 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 la cláusula ROW FILTER.
- FLOW
  
  Opcionalmente, define un flujo alineado con la creación de la tabla. Un flujo es una consulta con estado que actualiza el contenido de la tabla. Si FLOW no se especifica , puede usar AS query en su lugar o definir flujos por separado con CREATE FLOW. Puede especificar uno de los siguientes tipos de flujo:
  - INSERT POR NOMBRE
    
    Inserta datos en la tabla por nombre de columna. Si no se proporciona la ONCE opción , la 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.
    Nota:
    
    FLOW INSERT BY NAME es equivalente al uso de AS query. Las dos instrucciones siguientes tienen un comportamiento idéntico:
```
CREATE OR REFRESH STREAMING TABLE raw_data
AS SELECT * FROM STREAM read_files('abfss://my_path');

CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');
```
  - Una vez
    
    Opcionalmente, define el flujo como un flujo de un solo uso, como un reposición. Cuando ONCE se proporciona, la consulta no es una consulta de streaming y el flujo se ejecuta una vez de forma predeterminada. Si la tabla se actualiza con una actualización completa, el ONCE flujo se ejecuta de nuevo para volver a crear los datos. ONCE solo se aplica a INSERT BY NAME los flujos.
  - AUTO CDC
    
    Importante
    
    Disponible en Databricks Runtime 17.3 y versiones posteriores y en el PREVIEW canal pipelines.
    
    Define un AUTO CDC flujo que procesa los registros de captura de datos modificados (CDC) de un origen en la tabla. Use AUTO CDC cuando los datos de origen incluyan la semántica CDC. Consulte Las API DE AUTO CDC: Simplificación de la captura de datos modificados con canalizaciones.
  - Reemplazo WHERE de la consulta BY NAME del predicado
    
    Importante
    
    FLOW REPLACE WHERE está en beta. Requiere el uso del canal de vista previa de canalizaciones: establezca la pipelines.channel propiedad table en "PREVIEW".
    
    Define un REPLACE WHERE flujo que vuelve a calcular y sobrescribe solo las filas que coinciden predicate, dejando sin modificar todas las demás filas. Se usa REPLACE WHERE para el procesamiento por lotes incremental de combinaciones y agregaciones, datos de llegada tardía, evolución del esquema y reposición. BY NAME es obligatorio. Consulte Procesamiento por lotes con flujos REPLACEWHERE.
Consulta as

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 agregar la skipChangeCommits opción de lectura para controlar los 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.
- Opciones de lectura
  
  Puede especificar opciones de lectura en la consulta para configurar cómo se leen los datos desde el origen. Por ejemplo, puede especificar skipChangeCommits que omita las confirmaciones de cambios en los datos de origen. Las opciones de lectura se especifican como una asignación en la cláusula de consulta WITH . Por ejemplo:
```
SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS=TRUE, STARTINGVERSION=X)
```
  =TRUE es opcional, por lo que también puede especificar una opción booleana como esta:
```
SELECT * FROM STREAM source_table WITH (SKIPCHANGECOMMITS)
```
  Nota:
  
  Las opciones de lectura solo se admiten para Databricks Runtime 17.3 y versiones posteriores.
  
  Las siguientes opciones de lectura son compatibles con Delta, para obtener más información sobre cada opción, consulte Lecturas y escrituras de streaming de tablas de Delta Lake.
  - maxFilesPerTrigger
  - maxBytesPerTrigger
  - startingVersion
  - startingTimestamp
  - readChangeFeed
  - withEventTimeOrder
  - skipChangeCommits

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 instrucción CREATE OR REFRESH 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.

Examples

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

-- Use automatic liquid clustering to let Databricks choose the clustering columns:
CREATE OR REFRESH STREAMING TABLE customers_bronze_auto
CLUSTER BY AUTO
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/customers/*", format => "csv")

-- 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 with an identity column:
CREATE OR REFRESH STREAMING TABLE customers_with_id (
  customer_id BIGINT GENERATED ALWAYS AS IDENTITY,
  name string,
  region string
)
AS SELECT name, region FROM STREAM(customers_bronze)

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

-- Define a streaming table with an inline append flow:
CREATE OR REFRESH STREAMING TABLE raw_data
FLOW INSERT BY NAME SELECT * FROM STREAM read_files('abfss://my_path');

-- Define a streaming table with an inline AUTO CDC flow:
CREATE OR REFRESH STREAMING TABLE target
FLOW AUTO CDC
FROM stream(cdc_data.users)
KEYS (userId)
SEQUENCE BY sequenceNum
STORED AS SCD TYPE 1;

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-07-10