API-интерфейсы AUTO CDC: упрощение отслеживания измененных данных с помощью конвейеров

Конвейеры Lakeflow упрощают захват изменений данных (CDC) с помощью API AUTO CDC и AUTO CDC FROM SNAPSHOT. Эти API автоматизируют сложность вычисления медленно меняющихся размерностей (SCD) Типа 1 и Типа 2 либо из потока CDC, либо снимков базы данных, а AUTO CDC API также вычисляет двухвременную историю из потока CDC. Чтобы узнать больше о SCD типа 1, SCD типа 2 и битемпоральном, см. раздел Change data capture и snapshots.

Замечание

AUTO CDC API-интерфейсы заменяют APPLY CHANGES API и имеют тот же синтаксис. APPLY CHANGES API по-прежнему доступны, но Databricks рекомендует использовать AUTO CDC API на их месте.

Используемый API зависит от источника измененных данных:

  • AUTO CDC: Используйте это, когда в исходной базе данных включён канал CDC. AUTO CDC обрабатывает изменения из потока данных изменений (CDF). Поддерживается как в конвейерных интерфейсах SQL, так и в Python.
  • AUTO CDC FROM SNAPSHOT: Используйте это, когда CDC не включён в исходной базе данных и доступны только снимки. Этот API сначала сравнивает моментальные снимки, чтобы определить изменения, а затем обрабатывает их. Поддерживается как в конвейерных интерфейсах SQL, так и в Python.

Для объектов SCD типа 1 можно также объединить AUTO CDC FROM SNAPSHOT поток с одним или несколькими AUTO CDC потоками. Используйте этот паттерн, когда источник периодически предоставляет авторитетные снимки и CDC-поток с меньшей задержкой между снимками.

API поддерживают следующие типы хранения целевых таблиц:

  • SCD тип 1: Обновлять записи напрямую. История обновленных записей не сохраняется. Поддерживается обоими API.
  • SCD Тип 2: Сохранять историю записей — либо по всем обновлениям, либо по обновлениям определённого набора столбцов. Поддерживается обоими API.
  • Битемпоральный: Сохраняет историю записей в двух временных линиях: деловое время, когда изменение вступило в силу, и системное время, когда система его записала. Поддерживается только AUTO CDC. См. Bitemporal AUTO CDC.

AUTO CDC также поддерживает частичные обновления, когда запись изменений обновляет только часть столбцов. См. раздел "Применить частичные обновления".

AUTO CDC API не поддерживаются декларативными конвейерами Apache Spark.

Сведения о синтаксисе и других ссылках см. в AUTO CDC INTO (конвейеры), create_auto_cdc_flow и create_auto_cdc_from_snapshot_flow.

Замечание

Используйте эти API для обновления таблиц в ваших конвейерах на основе изменений исходных данных. Сведения о записи и запросе сведений об изменении на уровне строк для таблиц Delta см. в разделе "Использование канала изменений" в Azure Databricks.

Требования

Чтобы использовать API CDC, ваш конвейер должен быть настроен на использование бессерверных конвейеров Lakeflow или конвейеров Lakeflow Pro или Advancedредакций.

Как работает AUTO CDC

Чтобы выполнить обработку CDC с AUTO CDC, создайте потоковую таблицу, а затем используйте оператор AUTO CDC ... INTO в SQL или функцию create_auto_cdc_flow() в Python, чтобы указать источник, ключи и последовательности для канала изменений. Сведения о том, как работает логика упорядочивания и SCD, см. в разделе "Изменение данных и моментальных снимков". Ознакомьтесь с примерами AUTO CDC.

Для первоначальной гидратации из источника с каналом изменений используйте AUTO CDC с потоком once, а затем продолжайте обработку потока изменений. См. статью "Репликация внешней таблицы RDBMS" с помощью AUTO CDC.

Сведения о синтаксисе см. в разделе AUTO CDC INTO (конвейеры) или create_auto_cdc_flow.

Как работает AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT определяет изменения в исходных данных, сравнивая снимки по порядку. Моментальные снимки можно считывать непосредственно из таблицы Delta, файлов облачного хранилища или JDBC. Поддерживается как в конвейерных интерфейсах SQL, так и в Python.

Для выполнения обработки CDC с AUTO CDC FROM SNAPSHOT создайте потоковую таблицу, а затем определите поток:

  • В SQL используйте форму AUTO CDC ... FROM SNAPSHOT оператора CREATE FLOW или встроенное предложение FLOW AUTO CDC в CREATE STREAMING TABLE. Источник — это обязательное условие FROM SNAPSHOT (snapshot_query) и необязательное условие WITH VERSION (version_query), которое выбирает следующую версию снимка для обработки. См. CREATE FLOW (pipelines).
  • В Python используйте функцию create_auto_cdc_from_snapshot_flow() для указания снимка, ключей и других аргументов. См. create_auto_cdc_from_snapshot_flow.

Дополнительные сведения о двух шаблонах приема данных и их использовании см. в разделе "Шаблоны обработки моментальных снимков". Ознакомьтесь с примерами AUTO CDC FROM SNAPSHOT.

Объедините AUTO CDC FROM SNAPSHOT и AUTO CDC

Версия моментального снимка и столбец секвенирования CDC образуют одну область упорядочения. Событие CDC, более свежее последнего снимка, имеет приоритет над снимком. Более новый снимок имеет приоритет над старыми событиями CDC и рассматривает ключи, отсутствующие в снимке, как удаленные в версии снимка. Новые события CDC могут сохранять или восстанавливать эти ключи.

Единый целевой объект должен соответствовать следующим требованиям:

  • Целевой объект использует SCD Type 1.
  • Целевой объект имеет ровно один AUTO CDC FROM SNAPSHOT поток и один или несколько AUTO CDC потоков.
  • Каждый AUTO CDC поток имеет уникальное название.
  • Все потоки используют одинаковое количество ключей в одном и том же порядке. Имена ключей snapshot-flow сравниваются с именами ключей AUTO CDC без учета регистра. Несколько AUTO CDC потоков должны использовать одинаковые имена ключей и регистр.
  • Версия с снимком и каждый столбец секвенирования CDC имеют абсолютно одинаковый тип данных.
  • Поток AUTO CDC FROM SNAPSHOT не задаёт ожиданий.
  • Потоки не используют AUTO CDC или его варианты со списком столбцов.
  • Конвейер обработки использует триггерный режим.
  • Если вы добавляете AUTO CDC поток к существующей AUTO CDC FROM SNAPSHOT цели, цель не должна содержать пользовательские колонки, имена которых конфликтуют с зарезервированными AUTO CDC системными столбцами.

Оба потока AUTO CDC и AUTO CDC FROM SNAPSHOT могут использовать интерфейс конвейера Python или SQL. Вы можете смешивать SQL и процессы Python в одном назначении.

Используйте once=True в Python или ONCE в SQL в потоке AUTO CDC FROM SNAPSHOT, чтобы одноразово дозаполнить снимок, пока потоки AUTO CDC продолжают обрабатывать новые события при последующих обновлениях. Полное обновление целевого объекта повторно запускает процесс создания одноразового снимка. Для полного примера см. Добавить обратное заполнение в таблицу AUTO CDC SCD Type 1.

Этот паттерн не поддерживает SCD типа 2 или битемпоральные цели.

Используйте несколько столбцов для упорядочивания

Чтобы упорядочить по нескольким столбцам (например, метке времени и идентификатору для разрыва связей), используйте STRUCT, чтобы объединить их. API сортирует сначала по первому полю, а в случае равенства учитывает второе поле и так далее.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Питон

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

Примеры AUTO CDC

В следующих примерах показана обработка данных SCD типов 1 и 2 с использованием источника данных о изменениях. Пример данных создает новые записи пользователей, удаляет запись пользователя и обновляет записи пользователей. В примере SCD Type 1 последние UPDATE операции приходят поздно и удаляются из целевой таблицы, иллюстрируя обработку событий в неправильном порядке.

Ниже приведены входные записи, используемые в этих примерах. Эти данные создаются путем выполнения запроса в разделе "Создание примера данных ".

userId имя city Операция номер последовательности
124 Рауль Оахака INSERT 1
123 Isabel Монтеррей INSERT 1
125 Мерседес Тихуана INSERT 2
126 Лилия Канкун INSERT 2
123 ноль ноль Удалить 6
125 Мерседес Guadalajara UPDATE 6
125 Мерседес Мехикали UPDATE 5
123 Isabel Чиуауа UPDATE 5

Если вы раскомментируете окончательную строку в примере запроса создания данных, она добавляет следующую запись, которая указывает на необходимость очистки таблицы в sequenceNum=3.

userId имя city Операция номер последовательности
ноль ноль ноль УКОРАЧИВАТЬ 3

Замечание

Все приведенные ниже примеры включают параметры для указания обоих DELETE и TRUNCATE операций, но каждый из них является необязательным.

Создание образцов данных

Выполните следующие инструкции, чтобы создать образец набора данных. Этот код не предназначен для запуска в рамках определения конвейера. Запустите его из папки исследования конвейера, а не из папки преобразований.

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

Обработка обновлений SCD Type 1

SCD Type 1 сохраняет только последнюю версию каждой записи. В следующем примере считывается из потока данных изменений, созданного выше, и применяются изменения к целевой потоковой таблице. Для запуска этого кода нужен конвейер. См. статью "Разработка и отладка конвейеров ETL" с помощью редактора конвейеров Lakeflow.

Питон

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;

После выполнения примера SCD Type 1 целевая таблица содержит следующие записи:

userId имя city
124 Рауль Оахака
125 Мерседес Guadalajara
126 Лилия Канкун

Пользователь 123 (Isabel) был удален и не отображается. Пользователь 125 (Mercedes) показывает только последний город (Guadalajara), так как SCD Type 1 перезаписывает предыдущие значения. Ранее UPDATE на sequenceNum=5 было отменено, потому что поступило позднее обновление на sequenceNum=6.

После выполнения примера с раскомментированной записью TRUNCATE, таблица очищается в sequenceNum=3. Это означает, что записи 124 и 126 не находятся в таблице, а окончательная целевая таблица содержит только следующую запись:

userId имя city
125 Мерседес Guadalajara

Обработка обновлений SCD Type 2

SCD Type 2 сохраняет полный журнал изменений, создавая новые строки для каждой версии записи, а __START_AT__END_AT также столбцы, указывающие, когда каждая версия была активной.

Питон

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;

После запуска примера SCD Type 2 целевая таблица содержит следующие записи:

userId имя city __НАЧАТЬ_С __END_AT
123 Isabel Монтеррей 1 5
123 Isabel Чиуауа 5 6
124 Рауль Оахака 1 ноль
125 Мерседес Тихуана 2 5
125 Мерседес Мехикали 5 6
125 Мерседес Guadalajara 6 ноль
126 Лилия Канкун 2 ноль

Таблица сохраняет полную историю. Пользователь 123 имеет две версии (заканчиваются последовательностью 6 при удалении). Пользователь 125 имеет три версии, показывающие изменения города. Записи с __END_AT = null в настоящее время активны.

Отслеживание подмножества столбцов с помощью SCD Type 2

По умолчанию SCD Type 2 создает новую версию при каждом изменении значения столбца. Можно указать подмножество столбцов для отслеживания, чтобы изменения других столбцов обновляли текущую версию вместо создания новой записи журнала.

Пример, который следует, исключает city столбец из отслеживания истории.

Питон

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)

Так как city изменения не отслеживаются, обновления города перезаписывают текущую строку вместо создания новой версии. Целевая таблица содержит следующие записи:

userId имя city __НАЧАТЬ_С __END_AT
123 Isabel Чиуауа 1 6
124 Рауль Оахака 1 ноль
125 Мерседес Guadalajara 2 ноль
126 Лилия Канкун 2 ноль

Обрабатывайте двухвременные обновления

Бивременные таблицы отслеживают, когда каждое изменение вступило в силу (деловое время) и когда система его записала (системное время). Используйте SEQUENCE BY для столбца бизнес-времени и SYSTEM SEQUENCE BY (system_sequence_byна Python) для столбца системного времени. Битемпоральные цели требуют и того, и другого. Следующие фрагменты показывают только определение потока. Полный код конвейера см. Пример: Отслеживание ролей сотрудников с помощью двухвременного AUTO CDC.

Питон

dp.create_auto_cdc_flow(
  target = "employees_bitemporal",
  source = "employees",
  keys = ["employeeId"],
  sequence_by = "effectiveDate",
  system_sequence_by = "recordedAt",
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation"],
  stored_as_scd_type = "bitemporal"
)

SQL

CREATE FLOW apply_cdc AS AUTO CDC INTO employees_bitemporal
FROM stream(main.cdc_tutorial.employees_cdf)
KEYS (employeeId)
APPLY AS DELETE WHEN operation = "DELETE"
SEQUENCE BY effectiveDate
SYSTEM SEQUENCE BY recordedAt
COLUMNS * EXCEPT (operation)
STORED AS BITEMPORAL;

О столбцах, которые добавляют битемпоральные целевые таблицы, а также подробное пошаговое руководство с примерами данных, обновлениями не по порядку, исправлениями и запросами см. в Bitemporal AUTO CDC.

Примеры AUTO CDC FROM SNAPSHOT

В следующих разделах приведены примеры использования AUTO CDC FROM SNAPSHOT для обработки моментальных снимков в целевых таблицах SCD Type 1 или Type 2. Информацию о том, когда использовать этот API, см. в разделе Отслеживание изменений данных и снимки данных.

Примеры в этом разделе включают реализации Python и SQL. Полный синтаксис SQL см. CREATE FLOW (pipelines).

Пример: Обработка периодических снимков

Используйте этот подход, когда исходная система экспортирует полный снимок по регулярному графику, например, по одному файлу в час, и снимки приходят по порядку. В этом примере каждый снимок — это CSV-файл, название которого содержит дату и час экспорта, например users_2026-10-15-09.csv.

  • Реализация SQL использует WITH VERSION запрос для поиска следующего файла снимка и использует дату и час в названии файла в качестве версии снимка. Конвейер отслеживает последнюю обработанную версию.
  • Реализация Python использует функцию версии для поиска следующего файла снимка и использует дату и час в имени файла как версию снимка.

Вы также можете читать снимки с других типов исходных кодов, например, таблицу на Python или SQL, или соединение JDBC в Python. См. Инициализируйте пустую цель из одного снимка.

Шаг 1: Создайте файлы снимков-примеров.

Создайте том и запишите в него два почасовых файла снимков. Запустите следующий код из блокнота, находящегося в папке explorations вашего пайплайна:

spark.sql("CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial")
spark.sql("CREATE VOLUME IF NOT EXISTS main.cdc_tutorial.hourly_snapshots")

dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-09.csv",
  "userId,city\n1,Oaxaca\n2,Monterrey\n3,Tijuana\n",
  overwrite=True
)

dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-10.csv",
  "userId,city\n2,Carmel\n3,Los Angeles\n4,Death Valley\n6,Kings Canyon\n",
  overwrite=True
)

Шаг 2. Запуск AUTO CDC FROM SNAPSHOT

Для запуска кода на этом этапе нужен конвейер. См. статью "Разработка и отладка конвейеров ETL" с помощью редактора конвейеров Lakeflow.

В конвейере создайте новый файл на Python или SQL в папкеtransformations, добавьте следующий код и запустите конвейер:

Питон

import re
from typing import Optional, Tuple

from pyspark import pipelines as dp
from pyspark.sql import DataFrame

snapshot_dir = "/Volumes/main/cdc_tutorial/hourly_snapshots/"
snapshot_pattern = re.compile(r"users_([0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2})[.]csv")

def get_next_snapshot(
  latest_snapshot_version: Optional[str]
) -> Optional[Tuple[DataFrame, str]]:
  snapshots = []

  for file in dbutils.fs.ls(snapshot_dir):
    match = snapshot_pattern.fullmatch(file.name)
    if match:
      version = match.group(1)
      if latest_snapshot_version is None or version > latest_snapshot_version:
        snapshots.append((version, file.path))

  if not snapshots:
    return None

  next_version, snapshot_path = min(snapshots)
  snapshot = (
    spark.read
      .format("csv")
      .option("header", True)
      .schema("userId INT, city STRING")
      .load(snapshot_path)
  )
  return (snapshot, next_version)

dp.create_streaming_table("target")

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

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
-- Read the snapshot file selected by WITH VERSION.
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/hourly_snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'users_',
    (SELECT version FROM current_snapshot_version()),
    '.csv'
  )
)
-- Select the earliest snapshot file that is newer than the last processed version.
-- The version is the fixed-width date and hour string in the file name.
WITH VERSION (
  WITH snapshot_files AS (
    SELECT regexp_extract(
      path,
      'users_([0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2})[.]csv$',
      1
    ) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/hourly_snapshots/')
    WHERE path RLIKE 'users_[0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{2}[.]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;

Запуск конвейера. Во время обновления поток обрабатывает снимок 09:00, за которым следует снимок с 10:00. AUTO CDC FROM SNAPSHOT сравнивает два снимка и обнаруживает, что пользователь 1 был удалён, пользователи 2 и 3 обновлены, а пользователи 4 и 6 были вставлены. Версии снимков, выделенные из имён файлов, хранятся в столбцах __START_AT и __END_AT.

При использовании SCD типа 2 целевая таблица содержит следующие записи:

userId city __НАЧАТЬ_С __END_AT
1 Оахака 2026-10-15-09 2026-10-15-10
2 Монтеррей 2026-10-15-09 2026-10-15-10
2 Кармель 15.10.2026 10:00 ноль
3 Тихуана 2026-10-15-09 2026-10-15-10
3 Лос-Анджелес 2026-10-15-10 ноль
4 Долина смерти 2026-10-15-10 ноль
6 Каньон Кингс 2026-10-15-10 ноль

Пользователь 1 был завершен (удален). Пользователи 2 и 3 имеют две версии, показывающие изменения города. Пользователи 4 и 6 были недавно добавлены.

Замечание

Чтобы использовать SCD Type 1 вместо SCD Type 2, установите stored_as_scd_type=1 в Python или STORED AS SCD TYPE 1 в SQL. Целевая таблица SCD Type 1 не содержит столбцов __START_AT и __END_AT.

При SCD Type 1 целевая таблица показывает только текущее состояние:

userId city
2 Кармель
3 Лос-Анджелес
4 Долина смерти
6 Каньон Кингс

Шаг 3: Смоделируйте новый моментальный снимок и выполните повторный запуск

Запишите ещё один файл с часовым снимком, чтобы имитировать прибытие нового экспорта. Запустите этот код из блокнота в папке explorations вашего пайплайна:

dbutils.fs.put(
  "/Volumes/main/cdc_tutorial/hourly_snapshots/users_2026-10-15-11.csv",
  "userId,city\n2,Carmel\n3,San Diego\n4,Death Valley\n6,Kings Canyon\n7,Joshua Tree\n",
  overwrite=True
)

Повторный запуск конвейера. Поток выбирает только снимок 11:00, потому что предыдущие версии уже были обработаны. Пользователь 3 обновлен, пользователь 7 вставлен, а остальные записи остаются без изменений.

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

Пример: Инициализация пустой цели из одного снимка

Используйте этот подход для инициализации нового пустого целевого объекта из одного авторитетного снимка. Установите once=True в Python или используйте ключевое слово ONCE в SQL, чтобы сделать это одноразовым процессом. После успешной фиксации потока последующие инкрементальные обновления конвейера пропускают его. Полное обновление очищает целевой объект и запускает поток снова.

В этом примере таблица используется в качестве источника снимка, а текущее состояние хранится с SCD Type 1. Чтобы добавить снимок в цель, которая также получает события из потока AUTO CDC, см. раздел «Добавить обратную загрузку в таблицу AUTO CDC SCD Type 1».

Шаг 1: Создайте исходный снимок

Создайте таблицу, содержащую полное текущее состояние источника:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE OR REPLACE TABLE main.cdc_tutorial.initial_users (
  userId INT,
  city STRING
);

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

Шаг 2: Создайте поток создания разового снимка

В конвейере создайте файл на Python или SQL в папкеtransformations, добавьте следующий код и запустите конвейер. См. статью "Разработка и отладка конвейеров ETL" с помощью редактора конвейеров Lakeflow.

Питон

from pyspark import pipelines as dp

dp.create_streaming_table("users")

dp.create_auto_cdc_from_snapshot_flow(
  target = "users",
  source = "main.cdc_tutorial.initial_users",
  keys = ["userId"],
  stored_as_scd_type = 1,
  once = True
)

SQL

CREATE OR REFRESH STREAMING TABLE users;

CREATE FLOW initial_users_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.initial_users
)
KEYS (userId)
STORED AS SCD TYPE 1;
Используйте источник JDBC в Python

Чтобы инициализировать целевой объект из JDBC-соединения на классических вычислениях, определите представление, которое читает исходную таблицу, и используйте имя представления в качестве source аргумента:

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

Устанавливайте source = "initial_users" и once = True когда создаёте поток.

После завершения обновления целевой объект содержит следующие строки:

userId city
1 Оахака
2 Монтеррей
3 Тихуана

Шаг 3: Проверьте поведение одноразового сценария

Замените исходные данные на другой снимок:

TRUNCATE TABLE main.cdc_tutorial.initial_users;

INSERT INTO main.cdc_tutorial.initial_users VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley');

Запустите ещё одно инкрементальное обновление конвейера. Целевой объект остаётся неизменным, потому что одноразовый поток данных уже завершился. Если вы запускаете полное обновление, пайплайн очищает целевой объект и снова запускает поток данных. Затем целевой объект содержит замещающий снимок.

Ограничения

  • Столбец последовательности должен быть сортируемым типом данных. NULL Значения секвенирования не поддерживаются.
  • Чтобы передавать данные в потоковом режиме из целевого объекта процесса AUTO CDC, читайте из его канала изменений. Дополнительные сведения см. в чтении потока изменяемых данных из целевой таблицы AUTO CDC.
  • Битемпоральные цели имеют следующие ограничения:
    • AUTO CDC FROM SNAPSHOT не поддерживается. Чтобы обработать снимки, сначала получите из них поток изменений. См. Обработка снимков с битемпоральным AUTO CDC.
    • Комбинированные AUTO CDC FROM SNAPSHOT и AUTO CDC потоки не поддерживаются.
    • Несколько AUTO CDC потоков, записывающих в одну и ту же битемпоральную цель, не поддерживаются.
    • TRACK HISTORY ON(track_history_column_listи track_history_except_column_list на Python) не поддерживается. Битемпоральные таблицы отслеживают историю каждого столбца.
    • COLUMNS TO UPDATE и IGNORE NULL UPDATES не поддерживаются.
    • APPLY AS TRUNCATE WHEN не поддерживается.

Дополнительные ресурсы

  • Последнее обновление: