CREATE STREAMING TABLE ... FLOW AUTO CDC

Se aplica a:casilla marcada como Sí Databricks SQL

Use la FLOW AUTO CDC cláusula con CREATE STREAMING TABLE para procesar registros de captura de datos modificados (CDC) de un origen en una tabla de streaming.

Anteriormente, la MERGE INTO instrucción se usaba normalmente para procesar registros CDC en Azure Databricks. Sin embargo, MERGE INTO puede producir resultados incorrectos debido a registros fuera de secuencia o requiere una lógica compleja para volver a ordenar los registros.

AUTO CDC simplifica cdc mediante el control automático de registros desordenados. Las claves se especifican para identificar registros, una columna de secuencia para ordenar y si se deben almacenar los resultados como scD tipo 1 (actualizaciones directas) o scD tipo 2 (seguimiento del historial).

Sintaxis

CREATE OR REFRESH STREAMING TABLE table_name
FLOW AUTO CDC
FROM source
KEYS (keys)
[IGNORE NULL UPDATES [ON {columnList | * EXCEPT (exceptColumnList)}]]
[APPLY AS DELETE WHEN condition]
[APPLY AS TRUNCATE WHEN condition]
SEQUENCE BY orderByColumn
[SYSTEM SEQUENCE BY systemOrderByColumn]
[COLUMNS {columnList | * EXCEPT (exceptColumnList)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2 | BITEMPORAL}]
[TRACK HISTORY ON {columnList | * EXCEPT (exceptColumnList)}]
[COLUMNS TO UPDATE columnName]

El comportamiento predeterminado de INSERT los eventos y UPDATE es actualizar los eventos CDC del origen: actualice las filas de la tabla de destino que coincidan con las claves especificadas o inserte una nueva fila cuando no exista un registro coincidente en la tabla de destino. El control de eventos DELETE se puede especificar con la condición APPLY AS DELETE WHEN.

Parámetros

  • source

    Origen de los datos. El origen debe ser un origen 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 más información sobre los datos de streaming, consulte Transformación de datos con canalizaciones.

  • KEYS

    Columna o combinación de columnas que identifican de forma única una fila en los datos de origen. Los valores de estas columnas se usan para identificar qué eventos CDC se aplican a registros específicos de la tabla de destino.

    Para definir una combinación de columnas, use una lista separada por comas de columnas.

    Esta cláusula es necesaria.

  • IGNORE NULL UPDATES

    Permite la ingesta de actualizaciones que contienen un subconjunto de las columnas de destino. Cuando un evento CDC coincide con una fila existente y IGNORE NULL UPDATES se especifica, las columnas con un null valor conservan sus valores existentes en el destino. Esto también se aplica a las columnas anidadas con un valor null.

    Para las actualizaciones parciales, agregue una ON cláusula para controlar qué columnas omiten null los valores:

    • IGNORE NULL UPDATES ON columnList: solo las columnas enumeradas conservan sus valores existentes cuando el valor entrante es null. Todas las demás columnas aplican valores explícitos null .
    • IGNORE NULL UPDATES ON * EXCEPT (exceptColumnList): todas las columnas excepto las enumeradas conservan sus valores existentes cuando el valor entrante es null. Las columnas enumeradas aplican valores explícitos null .

    Para obtener más información, vea Aplicar actualizaciones parciales.

    Esta cláusula es opcional.

    El valor predeterminado es sobrescribir las columnas existentes con null valores.

  • APPLY AS DELETE WHEN

    Especifica cuándo se debe tratar un evento de CDC como un DELETE en lugar de un upsert.

    Para los orígenes de tipo SCD 2, para controlar los datos desordenados, la fila eliminada se conserva temporalmente como una lápida en la tabla Delta subyacente y se crea una vista en la metastore que filtra estas lápidas. El intervalo de retención se puede configurar con la pipelines.cdc.tombstoneGCThresholdInSecondspropiedad tabla.

    Esta cláusula es opcional.

  • APPLY AS TRUNCATE WHEN

    Especifica cuándo se debe tratar un evento CDC como una tabla TRUNCATEcompleta. Dado que esta cláusula desencadena un truncamiento completo de la tabla de destino, solo se debe usar para casos de uso específicos que requieren esta funcionalidad.

    La APPLY AS TRUNCATE WHEN cláusula solo se admite para el tipo SCD 1. El tipo SCD 2 no admite la operación de truncamiento.

    Esta cláusula es opcional.

  • SEQUENCE BY

    Nombre de columna que especifica el orden lógico de los eventos CDC en los datos de origen. El procesamiento de canalización usa esta secuenciación para gestionar los eventos de cambio que llegan fuera de orden.

    Si se necesitan varias columnas para la secuenciación, use una STRUCT expresión: ordenará primero por el primer campo struct, después por el segundo campo si hay un empate, etc.

    Las columnas especificadas deben ser tipos de datos ordenables.

    Esta cláusula es necesaria.

  • SYSTEM SEQUENCE BY

    Important

    Bitemporales AUTO CDC está en beta.

    Nombre de columna que especifica la hora del sistema en la que se conoce cada evento CDC para el sistema. Se usa con STORED AS BITEMPORAL para realizar un seguimiento de los cambios tanto en el tiempo empresarial (SEQUENCE BY) como en la hora del sistema. Consulte BItemporales AUTO CDC.

    Las columnas especificadas deben ser tipos de datos ordenables.

    Esta cláusula es opcional y solo se aplica a las tablas bitemporales.

  • COLUMNS

    Especifica un subconjunto de columnas que se van a incluir en la tabla de destino. Puede hacer lo siguiente:

    • Especifique la lista completa de columnas que se van a incluir: COLUMNS (userId, name, city).
    • Especifique una lista de columnas que se van a excluir: COLUMNS * EXCEPT (operation, sequenceNum)

    Esta cláusula es opcional.

    El valor predeterminado es incluir todas las columnas de la tabla de destino cuando no se especifica la COLUMNS cláusula .

  • STORED AS

    Si se van a almacenar registros como tipo SCD 1, tipo SCD 2 o bitemporales.

    Establézcalo en BITEMPORAL para realizar un seguimiento de los cambios tanto en la hora empresarial como en la hora del sistema. Bitemporales requiere SYSTEM SEQUENCE BY y está en Beta. Consulte BItemporales AUTO CDC.

    Esta cláusula es opcional.

    El valor predeterminado es SCD de tipo 1.

  • TRACK HISTORY ON

    Especifica un subconjunto de columnas de salida para generar registros de historial cuando hay cambios en esas columnas especificadas. Puede hacer lo siguiente:

    • Especifique la lista completa de columnas para realizar el seguimiento: COLUMNS (userId, name, city).
    • Especifique una lista de columnas que se excluirán del seguimiento: COLUMNS * EXCEPT (operation, sequenceNum)

    Esta cláusula es opcional. El valor predeterminado es realizar un seguimiento del historial de todas las columnas de salida cuando hay cambios, equivalentes a TRACK HISTORY ON *.

  • COLUMNS TO UPDATE

    Especifica el nombre de una columna de origen que contiene, para cada registro de cambio, el conjunto de columnas que se va a actualizar como una matriz de cadenas de nombre de columna (array<string>). Las columnas que no están en la matriz mantienen sus valores de destino existentes, mientras que las columnas enumeradas se escriben desde el origen, incluidos los valores explícitos null .

    Use esta cláusula para las actualizaciones parciales cuando cada registro de cambios actualice un conjunto diferente de columnas y tenga que aplicar valores explícitos null .

    No se puede usar COLUMNS TO UPDATE junto con IGNORE NULL UPDATESy no se admite para tablas bitemporales.

    Para obtener más información, vea Aplicar actualizaciones parciales.

    Esta cláusula es opcional.

Ejemplos

-- SCD type 1: apply CDC changes with direct updates (no history)
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  SEQUENCE BY sequenceNum
  STORED AS SCD TYPE 1;

-- SCD type 2: retain a history of changes, with delete handling
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2;

-- SCD type 2 with history tracking on specific columns
> CREATE OR REFRESH STREAMING TABLE target
  FLOW AUTO CDC
  FROM stream(cdc_data.users)
  KEYS (userId)
  APPLY AS DELETE WHEN operation = "DELETE"
  SEQUENCE BY sequenceNum
  COLUMNS * EXCEPT (operation, sequenceNum)
  STORED AS SCD TYPE 2
  TRACK HISTORY ON * EXCEPT (city);
  • Last updated on