create_auto_cdc_from_snapshot_flow

La create_auto_cdc_from_snapshot_flow función crea un flujo que usa la funcionalidad de captura de datos modificados (CDC) de canalizaciones de Lakeflow para procesar los datos de origen de las instantáneas de base de datos. Consulte Funcionamiento de AUTO CDC FROM SNAPSHOT.

Nota:

Esta función reemplaza a la función apply_changes_from_snapshot()anterior . Las dos funciones tienen la misma firma. Databricks recomienda actualizar para usar el nuevo nombre.

Importante

Debe tener una tabla de streaming de destino para esta operación. Para crear la tabla de destino necesaria, puede usar la función create_streaming_table().

Syntax

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Nota:

Para AUTO CDC FROM SNAPSHOT el procesamiento, el comportamiento predeterminado es insertar una nueva fila cuando un registro coincidente con las mismas claves no existe en el destino. Si existe un registro coincidente, solo se actualiza si alguno de los valores de la fila ha cambiado. Se eliminan las filas con claves presentes en el conjunto de destino, pero que ya no están presentes en el conjunto de origen.

Para obtener más información sobre el procesamiento CDC con instantáneas, consulte Las API de AUTO CDC: Simplificación de la captura de datos modificados con canalizaciones. Para ejemplos de uso de la create_auto_cdc_from_snapshot_flow() función, véanse los ejemplos de la ingestión periódica de instantáneas, la ingestión histórica de instantáneas y el relleno SCD Tipo 1 .

Parámetros

Parámetro Tipo Description
target str Obligatorio. Nombre de la tabla que se va a actualizar. Puede usar la función create_streaming_table() para crear la tabla de destino antes de ejecutar la create_auto_cdc_from_snapshot_flow() función.
source str o lambda function Obligatorio. Nombre de una tabla o vista para realizar instantáneas periódicamente o una función lambda de Python que devuelve el DataFrame de instantánea para su procesamiento y la versión de la instantánea. Consulte Implementación del source argumento.
keys list Obligatorio. Columna o combinación de columnas que identifican de forma única una fila en los datos de origen. Se usa para identificar qué eventos CDC se aplican a registros específicos de la tabla de destino. Puede especificar cualquiera de las siguientes opciones:
  • Una lista de cadenas: ["userId", "orderId"]
  • Lista de funciones de Spark SQL col() : [col("userId"), col("orderId"]. Los argumentos para col() funciones no pueden incluir calificadores. Por ejemplo, puede usar col(userId), pero no puede usar col(source.userId).
stored_as_scd_type str o int Si se van a almacenar registros como tipo SCD 1 o SCD tipo 2. 1 Establézcalo en para el tipo 1 de SCD o 2 para el tipo 2 de SCD. El valor predeterminado es SCD de tipo 1.
track_history_column_list o track_history_except_column_list list Subconjunto de columnas de salida a las que se va a realizar un seguimiento del historial en la tabla de destino. Use track_history_column_list para especificar la lista completa de columnas a las que se va a realizar el seguimiento. Use track_history_except_column_list para especificar las columnas que se excluirán del seguimiento. Puede declarar cualquier valor como una lista de cadenas o como funciones de Spark SQL col() :
  • track_history_column_list = ["userId", "name", "city"]
  • track_history_column_list = [col("userId"), col("name"), col("city")]
  • track_history_except_column_list = ["operation", "sequenceNum"]
  • track_history_except_column_list = [col("operation"), col("sequenceNum")

Los argumentos para col() funciones no pueden incluir calificadores. Por ejemplo, puede usar col(userId), pero no puede usar col(source.userId). El valor predeterminado es incluir todas las columnas de la tabla de destino cuando no se pasa ningún track_history_column_list argumento o track_history_except_column_list a la función .
once bool Si ejecutar el flujo de instantáneas solo una vez. Cuando True, el flujo se omite durante actualizaciones incrementales posteriores después de que se confirma con éxito. Una actualización completa del objetivo vuelve a ejecutar el flujo. Utiliza esta opción para una sola carga de instantáneas o relleno. El valor predeterminado es False.

Notes

Para objetivos SCD Tipo 1 en pipelines activados, un create_auto_cdc_from_snapshot_flow() flujo puede compartir un destino con uno o más AUTO CDC flujos definidos en Python o SQL. Se aplican los siguientes requisitos:

  • Dale a cada AUTO CDC flujo un nombre único.
  • Usa el mismo número de tonalidades en el mismo orden para todos los flujos. Los nombres de claves de flujo de instantáneas se comparan de forma insensible a mayúsculas y minúsculas con AUTO CDC los nombres de clave. Los flujos múltiples AUTO CDC deben usar nombres clave y casing idénticos.
  • Utiliza exactamente el mismo tipo de datos para la versión de instantánea y la columna de secuenciación de cada AUTO CDC flujo.
  • No definas expectativas sobre el AUTO CDC FROM SNAPSHOT flujo.
  • No utilice IGNORE NULL UPDATES. En Python, no pongas ignore_null_updates, ignore_null_updates_column_list, ni ignore_null_updates_except_column_list en los create_auto_cdc_flow() flujos.
  • Si añades un AUTO CDC flujo 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.
  • No utilice este patrón para objetivos SCD Tipo 2 o bitemporales.

Por ejemplo, véase Añadir un relleno a una tabla AUTO CDC SCD Tipo 1.

Implementación del source argumento

La create_auto_cdc_from_snapshot_flow() función incluye el source argumento . Para procesar instantáneas históricas, se espera que el source argumento sea una función lambda de Python que devuelva dos valores a la create_auto_cdc_from_snapshot_flow() función: un DataFrame de Python que contiene los datos de instantánea que se van a procesar y una versión de instantánea.

La primera invocación de la función debe devolver una instantánea DataFrame y una versión. Si la primera invocación devuelve None, la actualización de la tubería falla. Después de procesar al menos una instantánea, vuelve None para indicar que no hay instantáneas adicionales disponibles.

A continuación se muestra la firma de la función lambda:

lambda Any => Optional[(DataFrame, Any)]
  • El argumento de la función lambda es la versión de instantánea procesada más recientemente.
  • El valor devuelto de la función lambda es None o una tupla de dos valores: el primer valor de la tupla es un DataFrame que contiene la instantánea que se va a procesar. El segundo valor de la tupla es el que corresponde a la versión de instantánea, la cual representa el orden lógico de la misma.

Un ejemplo que implementa y llama a la función lambda:

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

El entorno de ejecución de canalizaciones de Lakeflow realiza los pasos siguientes cada vez que se desencadena la canalización que contiene la create_auto_cdc_from_snapshot_flow() función:

  1. Ejecuta la next_snapshot_and_version función para cargar la siguiente instantánea DataFrame y la versión de instantánea correspondiente.
  2. Si la primera invocación no devuelve DataFrame, la actualización falla. Si una invocación posterior no devuelve DataFrame, la ejecución termina y la actualización de la tubería se marca como completada.
  3. Detecta los cambios en la nueva instantánea y los aplica incrementalmente a la tabla de destino.
  4. Devuelve al paso 1 para cargar la siguiente instantánea y su versión.
  • Última actualización el