create_auto_cdc_from_snapshot_flow

Функция create_auto_cdc_from_snapshot_flow создает поток, использующий конвейеры Lakeflow, функции отслеживания измененных данных (CDC) для обработки исходных данных из моментальных снимков базы данных. Узнайте , как работает AUTO CDC FROM SNAPSHOT.

Замечание

Эта функция заменяет предыдущую функцию apply_changes_from_snapshot(). Две функции имеют одинаковую сигнатуру. Databricks рекомендует обновиться, чтобы использовать новое название.

Это важно

Для этой операции должна быть целевая таблица потоковой передачи. Чтобы создать требуемую целевую таблицу, можно использовать функцию create_streaming_table().

Синтаксис

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
)

Замечание

Для AUTO CDC FROM SNAPSHOT обработки поведение по умолчанию заключается в вставке новой строки, если соответствующая запись с теми же ключами не существует в целевом объекте. Если соответствующая запись существует, она обновляется только в том случае, если все значения в строке изменились. Строки с ключами, которые присутствуют в целевом объекте, но больше не встречаются в источнике, удаляются.

Дополнительные сведения об обработке CDC с помощью моментальных снимков см. в статье API AUTO CDC: упрощение отслеживания измененных данных с помощью конвейеров. Для примеров использования create_auto_cdc_from_snapshot_flow() этой функции см. примеры периодического забора снимков, исторического введения снимков и заполнения SCD типа 1 .

Параметры

Параметр Тип Description
target str Обязательное. Имя обновляемой таблицы. Функцию create_streaming_table() можно использовать для создания целевой таблицы перед выполнением create_auto_cdc_from_snapshot_flow() функции.
source str или lambda function Обязательное. Имя таблицы или представления, для периодического создания моментального снимка, или лямбда-функция Python, которая возвращает обрабатываемый DataFrame моментального снимка и версию моментального снимка. См. раздел "Реализация аргументаsource".
keys list Обязательное. Столбец или сочетание столбцов, однозначно определяющих строку в исходных данных. Это позволяет определить, какие события CDC применяются к определенным записям в целевой таблице. Можно указать любой из следующих вариантов:
  • Список строк: ["userId", "orderId"]
  • Список функций Spark SQLcol(): [col("userId"), col("orderId"] Аргументы функций col() не могут включать квалификаторы. Например, можно использовать col(userId), но нельзя использовать col(source.userId).
stored_as_scd_type str или int Следует ли хранить записи как SCD типа 1 или SCD типа 2. 1 Установите значение для SCD типа 1 или 2 для SCD типа 2. Значение по умолчанию — SCD тип 1.
track_history_column_list или track_history_except_column_list list Подмножество выходных столбцов для отслеживания истории в целевой таблице. Используется track_history_column_list для указания полного списка столбцов для отслеживания. Используется track_history_except_column_list для указания столбцов, которые следует исключить из отслеживания. Можно объявить любое значение в виде списка строк или как функции 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")

Аргументы функций col() не могут включать квалификаторы. Например, можно использовать col(userId), но нельзя использовать col(source.userId). Значение по умолчанию — включать все столбцы в целевую таблицу, если в функцию не передаются аргументы track_history_column_list или track_history_except_column_list.
once bool Стоит ли запускать поток снимков только один раз. Когда True, поток пропускается при последующих инкрементальных обновлениях после успешного фиксирования. Полное обновление цели повторяет поток. Используйте этот вариант для одноразовой загрузки или заполнения снимка. Значение по умолчанию — False.

Notes

Для целей SCD типа 1 в триггерных конвейерах один create_auto_cdc_from_snapshot_flow() поток может делить цель с одним или несколькими AUTO CDC потоками, определёнными на Python или SQL. К нему предъявляются следующие требования.

  • Дайте каждому AUTO CDC потоку уникальное название.
  • Используйте одинаковое количество ключей в одном порядке для всех потоков. Имена ключей с потоком снимков сравниваются с ключевыми именами AUTO CDC без падежа. Несколько AUTO CDC потоков должны использовать одинаковые имена ключей и корпус.
  • Используйте точно один и тот же тип данных для версии снимка и столбца секвенирования каждого AUTO CDC потока.
  • Не определяйте ожидания по потоку AUTO CDC FROM SNAPSHOT .
  • Не используйте IGNORE NULL UPDATES. В Python не устанавливайте ignore_null_updates, ignore_null_updates_column_list, или ignore_null_updates_except_column_list на create_auto_cdc_flow() потоках.
  • Если вы добавляете AUTO CDC поток к существующей AUTO CDC FROM SNAPSHOT цели, цель не должна содержать пользовательские колонки, имена которых конфликтуют с зарезервированными AUTO CDC системными столбцами.
  • Не используйте этот паттерн для SCD типа 2 или битемпоральных целей.

Для примера см. Добавить заполнение в таблицу AUTO CDC SCD Type 1.

Реализация аргумента source

Функция create_auto_cdc_from_snapshot_flow() включает source аргумент. Для обработки исторических моментальных снимков ожидается, что аргумент source будет лямбда-функцией Python, которая возвращает два значения в функцию create_auto_cdc_from_snapshot_flow(): DataFrame Python, содержащий данные моментального снимка для обработки, и версию моментального снимка.

Первый вызов функции должен возвращать снимок DataFrame и версию. Если возвращается первое взыскание None, обновление конвейера не работает. После обработки хотя бы одного снимка возвращайтесь None , чтобы указать, что дополнительных снимков нет.

Ниже приведена подпись лямбда-функции:

lambda Any => Optional[(DataFrame, Any)]
  • Аргументом для лямбда-функции является наиболее недавно обработанная версия моментального снимка.
  • Возвращаемое значение лямбда-функции — None или кортеж из двух значений: первое значение кортежа — это DataFrame, содержащий моментальный снимок для обработки. Вторым значением кортежа является версия моментального снимка, которая обозначает его логический порядок.

Пример, реализующий и вызывающий лямбда-функцию:

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,
  # ...
)

Среда выполнения конвейеров Lakeflow выполняет следующие действия при каждом запуске конвейера, содержащего create_auto_cdc_from_snapshot_flow() функцию:

  1. next_snapshot_and_version Функция запускается, чтобы загрузить следующий DataFrame для моментального снимка и соответствующую версию моментального снимка.
  2. Если первый вызов не возвращает DataFrame, обновление не проходит. Если позднее вызов не возвращает DataFrame, запуск завершается, и обновление конвейера отмечается как завершённое.
  3. Обнаруживает изменения в новом моментальном снимке и пошагово применяет их к целевой таблице.
  4. Возвращается к шагу 1, чтобы загрузить следующий снимок и его версию.
  • Последнее обновление: