AUTO CDC APIs: Simplifica la captura de datos de cambios con canalizaciones

Las canalizaciones de Lakeflow simplifican la captura de datos de cambios (CDC) con las API AUTO CDC y AUTO CDC FROM SNAPSHOT. Estas API automatizan la complejidad de calcular dimensiones de cambio lento (SCD) de tipo 1 y tipo 2 a partir de una fuente CDC o de instantáneas de base de datos. La AUTO CDC API también admite el seguimiento bitemporal, que registra los cambios en dos dimensiones temporales (Beta). Para obtener más información sobre SCD de tipo 1 y tipo 2, consulte Captura de datos de cambios e instantáneas. Para obtener más información sobre el seguimiento bitemporal, consulte Bitemporal AUTO CDC.

Nota:

Las AUTO CDC API reemplazan las APPLY CHANGES API y tienen la misma sintaxis. Las APPLY CHANGES API siguen estando disponibles, pero Databricks recomienda usar las AUTO CDC API en su lugar.

La API que use depende del origen de los datos modificados:

  • AUTO CDC: Utilice esto cuando la base de datos de origen tenga una fuente CDC habilitada. AUTO CDC procesa los cambios de un flujo de datos de cambios (CDF). Se admite tanto en las interfaces SQL como Python de la canalización.
  • AUTO CDC FROM SNAPSHOT: Use esto cuando CDC no está habilitado en la base de datos de origen y solo hay instantáneas disponibles. Esta API compara las instantáneas para determinar los cambios y, a continuación, los procesa. Se admite tanto en las interfaces SQL como Python de la canalización.

Para objetivos SCD Tipo 1, también puedes combinar un AUTO CDC FROM SNAPSHOT flujo con uno o más AUTO CDC flujos. Utiliza este patrón cuando una fuente proporciona instantáneas de referencia periódicas y una señal CDC de menor latencia entre instantáneas.

Ambas API admiten la actualización de tablas mediante SCD Tipo 1 y Tipo 2:

  • Use SCD Type 1 para actualizar los registros directamente. El historial no se conserva para los registros actualizados.
  • Use SCD Type 2 para conservar un historial de registros, ya sea en todas las actualizaciones o en actualizaciones de un conjunto de columnas especificado.

Solo para AUTO CDC, también puede usar almacenamiento bitemporal, que amplía el historial de SCD de tipo 2 para realizar un seguimiento de los cambios en dos dimensiones temporales: tiempo de negocio y tiempo del sistema. Bitemporal está en beta. Consulte AUTO CDC bitemporal.

AUTO CDC también admite actualizaciones parciales, donde un registro de cambios solo actualiza un subconjunto de columnas. Consulte Aplicar actualizaciones parciales.

Las AUTO CDC API no son compatibles con las canalizaciones declarativas de Apache Spark.

Para consultar la sintaxis y otras referencias, consulte AUTO CDC INTO (pipelines), create_auto_cdc_flow y create_auto_cdc_from_snapshot_flow.

Nota:

En esta página se describe cómo actualizar las tablas de las canalizaciones en función de los cambios en los datos de origen. Para obtener información sobre cómo registrar y consultar información de cambio de nivel de fila para las tablas Delta, consulte Uso de la fuente de distribución de datos modificados en Azure Databricks.

Requisitos

Si quiere usar las API de CDC, su canalización debe estar configurada para usar canalizaciones de Lakeflow sin servidor o las ediciones de canalizaciones de Lakeflow Pro o Advanced.

Funcionamiento de AUTO CDC

Para realizar el procesamiento CDC con AUTO CDC, cree una tabla de streaming y, a continuación, use la instrucción AUTO CDC ... INTO en SQL o la función create_auto_cdc_flow() en Python para especificar el origen, las claves y la secuenciación para el feed de cambios. Para una explicación de cómo funcionan la secuenciación y la lógica SCD, consulte Captura de datos modificados e instantáneas. Consulte los ejemplos de AUTO CDC.

Para la hidratación inicial desde un origen con una fuente de cambios, use AUTO CDC con un flujo once y, después, continúe procesando la fuente de cambios. Consulte Replicación de una tabla RDBMS externa mediante AUTO CDC.

Para más información sobre la sintaxis, consulte AUTO CDC INTO (pipelines) o create_auto_cdc_flow.

Funcionamiento de AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT determina los cambios en los datos de origen comparando instantáneas en orden. Puede leer instantáneas de una tabla Delta, archivos de almacenamiento en la nube o JDBC directamente. Se admite tanto en las interfaces SQL como Python de la canalización.

Para realizar el procesamiento CDC con AUTO CDC FROM SNAPSHOT, crea una tabla de streaming y luego define el flujo:

  • En SQL, se utiliza la AUTO CDC ... FROM SNAPSHOT forma de la CREATE FLOW sentencia o la cláusula incrustada FLOW AUTO CDC de CREATE STREAMING TABLE. La fuente es una cláusula obligatoria FROM SNAPSHOT (snapshot_query) y una cláusula opcional WITH VERSION (version_query) que selecciona la siguiente versión snapshot a procesar. Véase CREATE FLOW (pipelines).
  • En Python, usa la create_auto_cdc_from_snapshot_flow() función para especificar la instantánea, las claves y otros argumentos. Véase create_auto_cdc_from_snapshot_flow.

Para más información sobre los dos patrones de ingesta y cuándo usar cada uno, consulte Patrones de procesamiento de instantáneas. Consulte los ejemplos de AUTO CDC FROM SNAPSHOT.

Combina el AUTO CDC FROM SNAPSHOT y el AUTO CDC

La versión de instantánea y la columna de secuenciación de CDC forman un dominio de ordenación. Un evento del CDC más reciente que la última instantánea tiene prioridad sobre la instantánea. Una instantánea más reciente tiene prioridad sobre los eventos antiguos del CDC y trata las claves ausentes de la instantánea como eliminadas en la versión de la instantánea. Los eventos más recientes del CDC pueden preservar o restaurar esas claves.

Un objetivo unificado debe cumplir los siguientes requisitos:

  • El destino utiliza SCD Tipo 1.
  • El objetivo tiene exactamente un flujo AUTO CDC FROM SNAPSHOT y uno o más flujos AUTO CDC.
  • Cada AUTO CDC flujo tiene un nombre único.
  • Todos los flujos usan el mismo número de claves en el mismo orden. Los nombres de claves de flujo de instantáneas se comparan de forma insensible a mayúsculas y minúsculas con los nombres de claves AUTO CDC. Los flujos múltiples AUTO CDC deben usar nombres de clave y casing idénticos.
  • La versión snapshot y cada columna de secuenciación de CDC tienen exactamente el mismo tipo de datos.
  • El AUTO CDC FROM SNAPSHOT flujo no define las expectativas.
  • Los AUTO CDC flujos no utilizan IGNORE NULL UPDATES o sus variantes de lista de columnas.
  • La canalización utiliza el modo activado por desencadenador.
  • Si añades un flujo AUTO CDC a un destino existente AUTO CDC FROM SNAPSHOT, el destino no debe contener columnas de usuario cuyos nombres entren en conflicto con las columnas reservadas AUTO CDC del sistema.

Tanto AUTO CDC como AUTO CDC FROM SNAPSHOT pueden usar la interfaz de canalización de Python o SQL. Puedes mezclar flujos SQL y Python en el mismo destino.

Usa once=True en Python o ONCE en SQL en el flujo AUTO CDC FROM SNAPSHOT para rellenar una instantánea una vez mientras los AUTO CDC flujos continúan procesando nuevos eventos en actualizaciones posteriores. Una actualización completa del objetivo vuelve a ejecutar el flujo de instantáneas de una sola vez. Para un ejemplo completo, véase Añadir un backfill a una tabla AUTO CDC SCD Tipo 1.

Este patrón no soporta destinos SCD Tipo 2 ni bitemporales.

Uso de varias columnas para la secuenciación

Para secuenciar por varias columnas, por ejemplo, una marca de tiempo y un identificador para resolver empates, use un STRUCT para combinarlas. La API ordena primero por el primer campo y, en caso de empate, tiene en cuenta el segundo campo, y así sucesivamente.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Pitón

sequence_by = struct("timestamp_col", "id_col")

Ejemplos de AUTO CDC

En los ejemplos siguientes se muestra el procesamiento SCD Tipo 1 y Tipo 2 mediante una fuente de alimentación de datos de cambios. Los datos de ejemplo crean nuevos registros de usuario, eliminan un registro de usuario y actualizan los registros de usuario. En el ejemplo de SCD de tipo 1, las últimas operaciones UPDATE llegan tarde y se descartan de la tabla de destino, lo que muestra el control de eventos fuera de orden.

A continuación se muestran los registros de entrada usados en estos ejemplos. Estos datos se crean mediante la ejecución de la consulta en la sección Crear datos de ejemplo .

userId nombre city operation sequenceNum
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Lily Cancún INSERT 2
123 nulo nulo DELETE 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Si quita la marca de comentario de la fila final en la consulta de generación de datos de ejemplo, se inserta el siguiente registro, que especifica que se debe truncar la tabla, es decir, vaciarla, en sequenceNum=3:

userId nombre city operation sequenceNum
nulo nulo nulo TRUNCAR 3

Nota:

Todos los ejemplos siguientes incluyen opciones para especificar las DELETE operaciones y TRUNCATE , pero cada una es opcional.

Creación de datos de ejemplo

Ejecute las instrucciones siguientes para crear un conjunto de datos de ejemplo. Este código no está pensado para ejecutarse como parte de una definición de canalización. Ejecútelo desde la carpeta de exploración de la canalización, en lugar de la carpeta de transformaciones.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Procesar actualizaciones del tipo 1 de SCD

SCD Type 1 mantiene solo la versión más reciente de cada registro. El ejemplo siguiente lee desde la fuente de datos modificados creada anteriormente y aplica los cambios a una tabla de streaming de destino. Necesitas una tubería para ejecutar este código. Vea Cómo desarrollar y depurar canalizaciones de ETL con el Editor de canalizaciones de Lakeflow.

Pitón

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Después de ejecutar el ejemplo de SCD Type 1, la tabla de destino contiene los siguientes registros:

userId nombre city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Lily Cancún

Se eliminó el usuario 123 (Isabel) y no aparece. El usuario 125 (Mercedes) muestra solo la ciudad más reciente (Guadalajara) porque SCD Type 1 sobrescribe los valores anteriores. El UPDATE anterior en sequenceNum=5 se eliminó porque llegó una actualización posterior en sequenceNum=6.

Después de ejecutar el ejemplo con el registro TRUNCATE sin comentar, la tabla se borra en sequenceNum=3. Esto significa que los registros 124 y 126 no están en la tabla y la tabla de destino final contiene solo el registro siguiente:

userId nombre city
125 Mercedes Guadalajara

Procesar actualizaciones del tipo 2 de SCD

ScD Type 2 conserva un historial completo de cambios mediante la creación de nuevas filas para cada versión de un registro, con __START_AT columnas y __END_AT que indican cuándo estaba activa cada versión.

Pitón

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Después de ejecutar el ejemplo scD type 2, la tabla de destino contiene los siguientes registros:

userId nombre city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 nulo
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 nulo
126 Lily Cancún 2 nulo

La tabla conserva el historial completo. El usuario 123 tiene dos versiones (finalizó en la secuencia 6 cuando se eliminó). El usuario 125 tiene tres versiones que muestran los cambios en la ciudad. Los registros con __END_AT = null están activos actualmente.

Rastrear un subconjunto de columnas con SCD Type 2

De forma predeterminada, SCD Type 2 crea una nueva versión cada vez que cambia cualquier valor de columna. Puede especificar un subconjunto de columnas para realizar un seguimiento, de modo que los cambios realizados en otras columnas actualicen la versión actual en lugar de generar un nuevo registro de historial.

En el ejemplo siguiente se excluye la city columna del seguimiento del historial:

Pitón

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
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)

Como no se realiza un seguimiento de los cambios en city, las actualizaciones de la ciudad sobrescriben la fila existente en lugar de crear una nueva versión. La tabla de destino contiene los siguientes registros:

userId nombre city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 nulo
125 Mercedes Guadalajara 2 nulo
126 Lily Cancún 2 nulo

Ejemplos de AUTO CDC FROM SNAPSHOT

En las secciones siguientes se proporcionan ejemplos de cómo usar AUTO CDC FROM SNAPSHOT para procesar instantáneas en tablas de destino de Tipo 1 o Tipo 2 de SCD. Para obtener información contextual sobre cuándo usar esta API, consulte Captura de datos modificados e instantáneas.

Los ejemplos de esta sección incluyen implementaciones de Python y SQL. Para la sintaxis completa de SQL, véase CREATE FLOW (pipelines).

Ejemplo: Procesamiento de instantáneas periódicas

Utiliza este enfoque cuando las instantáneas lleguen regularmente y en orden. La interfaz de Python utiliza el orden de actualización del pipeline para el control de versiones. Para SQL, el ejemplo utiliza una tabla de versiones separada para identificar cuándo está disponible una nueva instantánea.

Puede leer instantáneas de varios tipos de origen, incluidas tablas Delta, archivos de almacenamiento en la nube y conexiones JDBC.

Paso 1: Crear datos de ejemplo

Cree una tabla que contenga datos de instantáneas. Ejecute el código siguiente desde un cuaderno o desde Databricks SQL en la carpeta explorations de su canalización:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Paso 2: Ejecute AUTO CDC FROM SNAPSHOT

Necesitas una tubería para ejecutar el código en este paso. Vea Cómo desarrollar y depurar canalizaciones de ETL con el Editor de canalizaciones de Lakeflow.

Para Python, elige un tipo fuente para la vista de instantáneas (el código de creación de ejemplo genera una tabla Delta):

Opción A: Leer desde una tabla Delta 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
Opción B: Leer desde el almacenamiento en la nube 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
Opción C: Lectura desde JDBC (solo proceso clásico) 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Luego añade la tabla objetivo y el flujo. La implementación de SQL utiliza la snapshot_version tabla para identificar cuándo hay disponible una nueva instantánea:

Pitón

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Después de la primera ejecución de la canalización, todos los registros se insertan como filas activas:

userId city __START_AT __END_AT
1 Oaxaca 0 nulo
2 Monterrey 0 nulo
3 Tijuana 0 nulo

Nota:

Para usar SCD Tipo 1 en su lugar y mantener solo el estado actual, configura stored_as_scd_type=1 en Python o STORED AS SCD TYPE 1 en SQL. En este caso, la tabla de destino no incluye ni __START_AT ni __END_AT columnas.

Paso 3: Simular una nueva instantánea y volver a ejecutar

Actualiza la tabla fuente para simular la llegada de una nueva instantánea (ejecuta este código desde un cuaderno o archivo SQL en la explorations carpeta de tu pipeline):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Repetición de la ejecución de la canalización AUTO CDC FROM SNAPSHOT compara la nueva instantánea con la anterior y detecta que se eliminó el usuario 1, se actualizaron los usuarios 2 y 3 y se insertaron los usuarios 4 y 6. Esto genera una fuente de cambios y usa AUTO CDC para crear la tabla de salida.

Después de la segunda ejecución con SCD Type 2, la tabla de destino contiene los siguientes registros:

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 nulo
3 Tijuana 0 1
3 Los Ángeles 1 nulo
4 Valle de la muerte 1 nulo
6 Cañón Kings 1 nulo

El usuario 1 finalizó (se eliminó). Los usuarios 2 y 3 tienen dos versiones que muestran sus cambios en la ciudad. Los usuarios 4 y 6 se insertaron recientemente.

Después de la segunda ejecución con SCD Type 1, la tabla de destino solo muestra el estado actual:

userId city
2 Carmel
3 Los Ángeles
4 Valle de la muerte
6 Cañón Kings

Ejemplo: Procesamiento de instantáneas mediante funciones de versión

Use este enfoque cuando necesite un control explícito sobre el orden de instantáneas. Por ejemplo, use este enfoque cuando llegan varias instantáneas al mismo tiempo o las instantáneas llegan fuera de orden. Defines cómo seleccionar la siguiente instantánea y su número de versión. La API procesa instantáneas en orden ascendente de versiones:

  • Si hay varias instantáneas almacenadas, el selector de ejemplo procesa las instantáneas disponibles en orden de versión.
  • Si una instantánea llega fuera de orden (por ejemplo, snapshot_3 llega después de snapshot_4), el selector de ejemplo la excluye porque su versión no es mayor que la última versión confirmada.
  • Si no hay nuevas instantáneas, la función de versión o la consulta no devuelve ningún resultado y no se produce ningún procesamiento.

Paso 1: Preparación de archivos de instantánea

Cree archivos CSV que contengan datos de instantánea y agréguelos a un volumen o a una ubicación de almacenamiento en la nube. Asigne un nombre cronológico a los archivos (por ejemplo, snapshot_1.csv, snapshot_2.csv).

Cada archivo debe contener columnas para userId y city. Por ejemplo:

snapshot_1.csv:

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId city
2 Carmel
3 Los Ángeles
4 Valle de la muerte

Paso 2: Ejecute AUTO CDC FROM SNAPSHOT con una función de versión

En una pipeline, crea un nuevo archivo Python o SQL en la transformations carpeta y añade el siguiente código. A continuación, ejecuta la canalización. Vea Cómo desarrollar y depurar canalizaciones de ETL con el Editor de canalizaciones de Lakeflow.

Pitón

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Nota:

Para usar SCD Tipo 1 en su lugar, configura stored_as_scd_type=1 en Python o STORED AS SCD TYPE 1 en SQL.

Después de procesar snapshot_1.csv, la tabla de destino contiene los siguientes registros:

userId city __START_AT __END_AT
1 Oaxaca 1 nulo
2 Monterrey 1 nulo
3 Tijuana 1 nulo

Después de procesar snapshot_2.csv, la tabla de destino contiene los siguientes registros:

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 nulo
3 Tijuana 1 2
3 Los Ángeles 2 nulo
4 Valle de la muerte 2 nulo

Nota:

Recuerde que, para SCD Type 1, la tabla es exactamente similar a la instantánea más reciente. La diferencia es que las consultas posteriores pueden usar el flujo de cambios para procesar solo los registros modificados.

Paso 3: Agregar nuevas instantáneas

Agregue un nuevo archivo CSV a la ubicación de almacenamiento con datos modificados (por ejemplo, valores de ciudad modificados, filas nuevas o filas eliminadas). Después, vuelva a ejecutar la canalización para procesar la nueva instantánea.

Limitaciones

Recursos adicionales

  • Se actualizó por última vez el