AUTO CDC API：使用管道简化变更数据捕获

Lakeflow 管道借助 AUTO CDC 和 AUTO CDC FROM SNAPSHOT API 简化了变更数据捕获（CDC）。这些 API 通过 CDC 源或数据库快照自动化计算渐变维度 (SCD) 类型 1 和类型 2 的复杂过程。该 AUTO CDC API 还支持双时间跟踪（Beta），可记录跨两个时间维度的变更。若要详细了解 SCD 类型 1 和类型 2，请参阅更改数据捕获和快照。有关双时态跟踪的详细信息，请参阅 Bitemporal AUTO CDC。

注释

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 和类型 2 更新表：

使用 SCD 类型 1 直接更新记录。不会保留已更新记录的历史记录。
使用 SCD 类型 2 保留有关所有更新或者对指定列集的更新的记录的历史记录。

仅对于 AUTO CDC，您还可以使用双时态存储，它将 SCD 2 型历史记录扩展为跨两个时间维度跟踪更改：业务时间和系统时间。 Bitemporal 目前为测试版。请参阅 Bitemporal AUTO CDC。

AUTO CDC 还支持部分更新，其中更改记录仅更新列的子集。请参阅 “应用部分更新”。

AUTO CDC Apache Spark 声明性管道不支持这些 API。

有关语法和其他引用，请参阅 AUTO CDC INTO（管道）、create_auto_cdc_flow 和 create_auto_cdc_from_snapshot_flow。

注释

利用这些 API 根据源数据的变化更新管道中的表。若要了解如何记录和查询 Delta 表的行级更改信息，请参阅在Azure Databricks上使用更改数据馈送。

要求

若要使用 CDC API，必须将您的管道配置为使用无服务器 Lakeflow 管道或 Lakeflow 管道 Pro 或 Advanced版本。

AUTO CDC 的工作原理

若要执行 AUTO CDC CDC 处理，请首先创建流式处理表，然后分别在 SQL 中使用 AUTO CDC ... INTO 语句或在 Python 中使用 create_auto_cdc_flow() 函数来指定变更跟踪源、关键字和排序。有关序列化和 SCD 逻辑的工作原理的说明，请参阅更改数据捕获和快照。请参阅 AUTO CDC 示例。

若要从具有更改源的源进行初始数据注入，请使用 AUTO CDC 和 once 流，然后继续处理更改源。请参阅使用 AUTO CDC 复制外部 RDBMS 表。

有关语法详细信息，请参阅 AUTO CDC INTO（管道）或 create_auto_cdc_flow。

AUTO CDC FROM SNAPSHOT 的运行机制

AUTO CDC FROM SNAPSHOT 通过比较顺序快照来确定源数据的更改。可以直接从 Delta 表、云存储文件或 JDBC 读取快照。管道 SQL 和 Python 接口都支持它。

要使用 AUTO CDC FROM SNAPSHOT 进行 CDC 处理，请创建一个流表，然后定义该流：

在SQL中，使用CREATE FLOW语句的AUTO CDC ... FROM SNAPSHOT形式或CREATE STREAMING TABLE的嵌入式FLOW AUTO CDC子句。源是必填的 FROM SNAPSHOT (snapshot_query) 子句和可选的 WITH VERSION (version_query) 子句，后者用于选择要处理的下一个快照版本。参见创建流程（管道）。
在 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 1 型。
目标恰好有一个 AUTO CDC FROM SNAPSHOT 流和一个或多个 AUTO CDC 流。
每个 AUTO CDC 流都有独特的名称。
所有流程使用相同数量的键，顺序相同。快照流键名与 AUTO CDC 键名比较时不区分大小写。多个 AUTO CDC 流程必须使用相同的密钥名和大小写。
快照版本和每个CDC排序顺序列的数据类型完全相同。
AUTO CDC FROM SNAPSHOT流程并未规定预期结果。
这些 AUTO CDC 流程不使用 IGNORE NULL UPDATES 或其列列表变体。
该管道使用触发模式。
如果你向现有AUTO CDC FROM SNAPSHOT目标添加AUTO CDC流程，目标中不得包含与保留AUTO CDC系统列名称冲突的用户列。

这两者AUTO CDC和AUTO CDC FROM SNAPSHOT流程都可以使用Python或SQL流水线接口。你可以在同一个目标中混合使用SQL和Python流程。

在 AUTO CDC FROM SNAPSHOT 流程中使用 Python 的 once=True 或 SQL 的 ONCE 来一次性回填快照，同时 AUTO CDC 流程在后续更新中继续处理新事件。目标的全面刷新会重新运行一次性快照流程。完整示例请参见 “向自动CDC SCD Type 1表添加回填”。

该模式不支持SCD第二型或双时态目标。

使用多个列进行排序

若要按多个列（例如时间戳和 ID 以解决冲突）进行排序，请使用STRUCT来组合它们。 API先按第一个字段进行排序，如果出现相同情况，则考虑第二个字段，以此类推。

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

AUTO CDC 示例

以下示例演示使用更改数据馈送源处理 SCD 类型 1 和类型 2。示例数据创建新的用户记录、删除用户记录和更新用户记录。在 SCD 类型 1 示例中，最后的 UPDATE 个操作由于延迟到达而被从目标表中删除，说明了无序事件处理。

下面是这些示例中使用的输入记录。通过在 “创建示例数据 ”部分中运行查询来创建此数据。

userId	姓名	city	操作	序列号
124	劳尔	瓦哈卡州	INSERT	1
123	伊莎贝尔	蒙特雷	INSERT	1
125	梅赛德斯	提华纳	INSERT	2
126	百合	坎昆	INSERT	2
123	null	null	DELETE	6
125	梅赛德斯	瓜达拉哈拉	UPDATE	6
125	梅赛德斯	Mexicali	UPDATE	5
123	伊莎贝尔	奇瓦瓦州	UPDATE	5

如果在示例数据生成查询中取消注释最后一行，它将插入以下记录，指定在 sequenceNum=3 位置进行表截断（清空表）：

userId	姓名	city	操作	序列号
null	null	null	截断	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 类型 1 更新

SCD 类型 1 仅保留每个记录的最新版本。以下示例从上面创建的更改数据馈送中读取数据，并将更改应用于流处理表目标。你需要一个流水线来运行这些代码。请参阅使用 Lakeflow 管道编辑器开发和调试 ETL 管道。

Python

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 类型 1 示例后，目标表包含以下记录：

userId	姓名	city
124	劳尔	瓦哈卡州
125	梅赛德斯	瓜达拉哈拉
126	百合	坎昆

用户 123 （Isabel）已被删除，并且未显示。用户 125 （梅赛德斯）仅显示最新的城市（瓜达拉哈拉），因为 SCD 类型 1 覆盖以前的值。之前在 UPDATE 的 sequenceNum=5 被删除，因为在 sequenceNum=6 的更新已经到来。

在运行未注释的 TRUNCATE 记录的示例后，表在 sequenceNum=3 处被清除。这意味着记录 124 和 126 不在表中，最终目标表仅包含以下记录：

userId	姓名	city
125	梅赛德斯	瓜达拉哈拉

处理 SCD 类型 2 更新

SCD 类型 2 通过为每个记录__START_AT__END_AT版本创建新行以及指示每个版本处于活动状态的列来保留更改的完整历史记录。

Python

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 类型 2 示例后，目标表包含以下记录：

userId	姓名	city	__START_AT	__END_AT
123	伊莎贝尔	蒙特雷	1	5
123	伊莎贝尔	奇瓦瓦州	5	6
124	劳尔	瓦哈卡州	1	null
125	梅赛德斯	提华纳	2	5
125	梅赛德斯	Mexicali	5	6
125	梅赛德斯	瓜达拉哈拉	6	null
126	百合	坎昆	2	null

该表保留完整的历史记录。用户 123 有两个版本（删除时结束于序列 6）。用户 125 有三个显示城市更改的版本。带 __END_AT = null 的记录当前处于活动状态。

使用 SCD 类型 2 跟踪列子集

默认情况下，SCD 类型 2 会在任何列值发生更改时创建新版本。可以指定要跟踪的列的子集，以便对其他列所做的更改将更新当前版本，而不是生成新的历史记录。

以下示例从历史记录跟踪中排除 city 列：

Python

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	__START_AT	__END_AT
123	伊莎贝尔	奇瓦瓦州	1	6
124	劳尔	瓦哈卡州	1	null
125	梅赛德斯	瓜达拉哈拉	2	null
126	百合	坎昆	2	null

AUTO CDC FROM SNAPSHOT 示例

以下部分提供了用于 AUTO CDC FROM SNAPSHOT 将快照处理到 SCD 类型 1 或类型 2 目标表的示例。有关何时使用此 API 的背景信息，请参阅更改数据捕获和快照。

本节示例包括Python和SQL实现。完整的SQL语法，请参见CREATE FLOW (pipelines)。

示例：处理周期性快照

当快照定期且有序地到达时，使用这种方法。 Python接口使用流水线更新顺序进行版本管理。对于 SQL 来说，示例使用一个独立的版本表来识别何时有新的快照可用。

可以从多个源类型（包括 Delta 表、云存储文件和 JDBC 连接）读取快照。

步骤 1：创建示例数据

创建包含快照数据的表。从管道的 explorations 文件夹中的笔记本或 Databricks SQL 运行以下代码：

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

步骤 2：运行 AUTO CDC FROM SNAPSHOT

你需要一个流水线来运行这一步的代码。请参阅使用 Lakeflow 管道编辑器开发和调试 ETL 管道。

对于 Python，选择快照视图的源类型（示例创建代码生成一个 Delta 表）：

选项 A：从 Delta 表读取

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")

选项 B：从云存储读取

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")

选项 C：从 JDBC 读取（仅限经典计算）

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

然后添加目标表和流程。 SQL 实现使用该 snapshot_version 表来识别何时有新的快照可用：

Python

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;

在第一次管道运行后，所有记录都作为活动行插入：

userId	city	__END_AT
1	瓦哈卡州	null
2	蒙特雷	null
3	提华纳	null

注释

改用SCD类型1，只保留当前状态，在Python中设置STORED AS SCD TYPE 1，或在SQL中设置stored_as_scd_type=1。在这种情况下，目标表不包含 __START_AT 和 __END_AT 列。

步骤 3：模拟新快照并重新运行

更新源表以模拟新的快照到来（在你的管道的 explorations 文件夹中的笔记本或 SQL 文件中运行此代码）：

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;

再次运行管道。 AUTO CDC FROM SNAPSHOT 将新快照与上一个快照进行比较，并检测到用户 1 已删除、用户 2 和 3 已更新，并插入了用户 4 和 6。这会生成一个更改提要，并使用 AUTO CDC 来创建输出表。

使用 SCD 类型 2 进行第二次运行后，目标表包含以下记录：

userId	city	__START_AT	__END_AT
1	瓦哈卡州	0	1
2	蒙特雷	0	1
2	卡梅尔	1	null
3	提华纳	0	1
3	洛杉矶	1	null
4	死亡谷	1	null
6	国王峡谷	1	null

用户 1 已结束（已删除）。用户 2 和 3 各有两个版本显示其城市更改。新插入了用户 4 和 6。

使用 SCD 类型 1 进行第二次运行后，目标表仅显示当前状态：

userId	city
2	卡梅尔
3	洛杉矶
4	死亡谷
6	国王峡谷

示例：使用版本函数处理快照

如果需要显式控制快照排序，请使用此方法。例如，当多个快照同时到达或快照无序到达时，请使用此方法。你定义如何选择下一个快照及其版本号。 API 按版本号升序处理快照：

如果存储中有多个快照，示例选择器会按版本顺序处理可用的快照。
如果快照是乱序到达的（例如，snapshot_3 在 snapshot_4 之后到达），示例选择器会排除它，因为其版本不大于上次提交的版本。
如果没有新的快照，版本函数或查询不会返回任何结果，也不会进行处理。

步骤 1：准备快照文件

创建包含快照数据的 CSV 文件，并将其添加到卷或云存储位置。按时间顺序命名文件（例如，snapshot_1.csvsnapshot_2.csv）。

每个文件应包含用于 userId 和 city 的列。例如：

snapshot_1.csv：

userId	city
1	瓦哈卡州
2	蒙特雷
3	提华纳

snapshot_2.csv：

userId	city
2	卡梅尔
3	洛杉矶
4	死亡谷

步骤 2：使用版本函数运行 AUTO CDC FROM SNAPSHOT

在管道中，在 transformations 文件夹中创建一个新的 Python 或 SQL 文件，并添加以下代码。然后运行管道。请参阅使用 Lakeflow 管道编辑器开发和调试 ETL 管道。

Python

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;

注释

如果想用 SCD Type 1，可以在 Python 中设置 STORED AS SCD TYPE 1，或在 SQL 中设置 stored_as_scd_type=1。

处理 snapshot_1.csv后，目标表包含以下记录：

userId	city	__START_AT	__END_AT
1	瓦哈卡州	1	null
2	蒙特雷	1	null
3	提华纳	1	null

处理 snapshot_2.csv后，目标表包含以下记录：

userId	city	__START_AT	__END_AT
1	瓦哈卡州	1	2
2	蒙特雷	1	2
2	卡梅尔	2	null
3	提华纳	1	2
3	洛杉矶	2	null
4	死亡谷	2	null

注释

请记住，对于 SCD 类型 1，表看起来与最新的快照完全相同。区别在于，下游查询可以使用更改源来仅处理更改的记录。

步骤 3：添加新快照

使用修改后的数据（例如，已更改的城市值、新行或删除行）将新的 CSV 文件添加到存储位置。然后再次运行管道以处理新快照。

局限性

排序列必须是可排序的数据类型。 NULL 不支持序列化值。
若要从 AUTO CDC 过程的目标端流式读取数据，请从其更改馈送中读取。有关详细信息，请参阅从 AUTO CDC 目标表读取变更数据流。

其他资源

更改数据捕获和快照：了解 CDC 概念、快照和 SCD 类型。
使用 AUTO CDC 复制外部 RDBMS 表：了解如何使用 once 流执行初始注入，然后继续处理更改。
用管道回填历史数据：学习如何回填仅追加和 AUTO CDC SCD Type 1 目标。
高级 AUTO CDC 主题：了解 AUTO CDC 目标的更改操作、读取更改数据馈送和处理指标。
倒带并重放管道：了解如何将 AUTO CDC 目标还原到较早的时间点并重新处理变更。
教程：使用变更数据捕获生成 ETL 管道

反馈

此页面是否有帮助？

上次更新时间为 2026-10-09