AUTO CDC API：使用管道簡化變更資料擷取

Lakeflow pipelines 透過 AUTO CDC 和 AUTO CDC FROM SNAPSHOT API 簡化了變更資料擷取（CDC）。這些 API 自動化計算來自 CDC 資料流或資料庫快照的類型 1 和類型 2 的緩慢變化維度（SCD）複雜度。 API AUTO CDC 也支援雙時序追蹤，能記錄跨兩個時間維度的變化（Beta）。欲了解更多關於 SCD Type 1 與 Type 2 的資訊，請參閱變更資料擷取與快照。欲了解更多雙時空追蹤，請參閱 Bitemporal AUTO CDC。

備註

API AUTO CDC 取代了 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 Type 1 目標，你也可以將一個 AUTO CDC FROM SNAPSHOT 流與一個或多個 AUTO CDC 流合併。當來源在快照間提供定期權威快照及低延遲的 CDC 資料流時，請使用此模式。

這兩個 API 都支援使用 SCD 類型 1 和類型 2 來更新資料表：

使用 SCD Type 1 直接更新紀錄。更新記錄不會保存歷史。
使用 SCD 類型 2 來保留記錄歷史，無論是所有更新還是指定欄位集合的更新。

僅適用於 AUTO CDC，你也可以使用雙時態儲存，將 SCD Type 2 的歷史追蹤擴展為可追蹤兩個時間維度上的變更：業務時間和系統時間。 Bitemporal 目前仍處於 Beta 版本。請參見 Bitemporal AUTO CDC。

AUTO CDC 也支援部分更新，即變更記錄只更新部分欄位。請參見「套用部分更新」。

Apache Spark 宣告式管線不支援這些 AUTO CDC API。

關於語法及其他參考，請參見 AUTO CDC INTO （pipelines）、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 CDConce流，然後繼續處理變更提要。請參考使用 AUTO CDC 複製外部 RDBMS 資料表。

語法細節請參見 AUTO CDC INTO （pipelines）或 create_auto_cdc_flow。

快照生成的自動CDC工作原理

AUTO CDC FROM SNAPSHOT 透過比較順序快照來判斷來源資料的變更。你可以直接從 Delta 表格、雲端儲存檔案或 JDBC 讀取快照。它同時支援於管線 SQL 與 Python 介面中。

若要使用 AUTO CDC FROM SNAPSHOT 進行 CDC 處理，建立串流表並定義流程：

在 SQL 中，使用 CREATE FLOW 陳述句的 AUTO CDC ... FROM SNAPSHOT 形式，或 FLOW AUTO CDC 的內嵌 CREATE STREAMING TABLE 子句。來源子句是一個必要 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 Type 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 流程上使用 once=True Python 或 ONCE SQL 一次性回填快照，讓 AUTO CDC 流程在後續更新時繼續處理新事件。目標的完整刷新會重新執行一次性快照流程。完整範例請參見「在 AUTO CDC SCD 類型 1 表格中新增回填」。

此模式不支援 SCD 第二型或雙時段目標。

使用多欄進行定序

若要依多個欄位排序（例如使用時間戳記和識別碼分辨前後順序），請使用 STRUCT 將它們合併。 API 會先依據第一個欄位排序，若出現平手則考慮第二個欄位，依此類推。

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

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

AUTO CDC 範例

以下範例展示了利用變更資料來源進行 SCD Type 1 與 Type 2 的處理。範例資料會建立新的使用者紀錄、刪除使用者記錄，並更新使用者紀錄。在 SCD Type 1 範例中，最後 UPDATE 的操作會延遲到達，並從目標表中移除，顯示出事件處理順序錯亂。

以下是這些範例中使用的輸入記錄。這些資料是透過在「建立樣本資料」區塊執行查詢而產生的。

userId	名稱	city	作業	序列編號
124	勞爾	瓦哈卡州	INSERT	1
123	Isabel	蒙特雷	INSERT	1
125	梅賽德斯	蒂華納	INSERT	2
126	莉莉	坎昆	INSERT	2
123	null	null	刪除	6
125	梅賽德斯	Guadalajara	UPDATE	6
125	梅賽德斯	墨西卡利	UPDATE	5
123	Isabel	Chihuahua	UPDATE	5

如果你在樣本資料生成查詢中取消註解最後一行，它會插入以下記錄，指定要在sequenceNum=3截斷該資料表（清除該資料表）。

userId	名稱	city	作業	序列編號
null	null	null	截斷	3

備註

下列所有範例都包含指定DELETE和TRUNCATE作業的選項，但每個選項都是可選的。

建立範例資料

執行以下語句以建立範例資料集。此程式碼並非設計用於管線定義的一部分。從管線的探索資料夾執行，而不是從 transformations 資料夾裡執行。

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 Type 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	梅賽德斯	Guadalajara
126	莉莉	坎昆

用戶 123（Isabel）已被刪除，未出現。使用者 125（Mercedes）只顯示最新的城市（瓜達拉哈拉），因為 SCD Type 1 會覆蓋先前的值。之前 UPDATE 的 at sequenceNum=5 被刪除，因為後來有更新 at sequenceNum=6 。

在執行範例並確保 TRUNCATE 記錄未被註解後，該表在 sequenceNum=3 會被清除。這表示記錄 124 和 126 不在表格中，最終目標表格僅包含以下記錄：

userId	名稱	city
125	梅賽德斯	Guadalajara

處理 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 Type 2 範例後，目標資料表包含以下記錄：

userId	名稱	city	__START_AT	__END_AT
123	Isabel	蒙特雷	1	5
123	Isabel	Chihuahua	5	6
124	勞爾	瓦哈卡州	1	null
125	梅賽德斯	蒂華納	2	5
125	梅賽德斯	墨西卡利	5	6
125	梅賽德斯	Guadalajara	6	null
126	莉莉	坎昆	2	null

這張桌子保存了完整的歷史。使用者 123 有兩個版本（刪除時止於序列 6）。用戶 125 有三個版本顯示城市變更。目前相關 __END_AT = null 紀錄仍在運作中。

用 SCD 類型 2 追蹤欄位子集

預設情況下，SCD Type 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	Isabel	Chihuahua	1	6
124	勞爾	瓦哈卡州	1	null
125	梅賽德斯	Guadalajara	2	null
126	莉莉	坎昆	2	null

快照中的自動CDC範例

以下章節提供使用 AUTO CDC FROM SNAPSHOT 將快照處理至 SCD 類型 1 或類型 2 目標資料表的範例。關於何時使用此 API 的背景，請參閱變更資料擷取與快照。

本節範例包括 Python 與 SQL 實作。完整 SQL 語法請參見 CREATE FLOW (pipelines)。

範例：處理週期性快照

當快照定期且有序地送達時，請使用此方法。 Python 介面使用管線更新順序來進行版本控制。以 SQL 為例，範例使用獨立的版本表來識別何時有新的快照可用。

你可以從多種來源類型讀取快照，包括 Delta 表格、雲端儲存檔案和 JDBC 連線。

步驟 1：建立範例資料

建立一個包含快照資料的資料表。請從你的管線資料夾中的筆記本或 Databricks SQL explorations 執行以下程式碼：

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：從快照中運行自動 CDC

你需要一個管線來執行這個步驟的程式碼。請參閱使用 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 Type 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 Type 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 Type 1 後，目標表僅顯示當前狀態：

userId	city
2	卡梅爾
3	洛杉磯
4	死亡谷
6	國王峽谷

範例：使用版本函數進行程序快照

當你需要明確控制快照排序時，請使用此方法。例如，當多個快照同時抵達或快照順序錯亂時，可以使用此方法。你定義如何選擇下一個快照及其版本號。 API 依版本依序從低到高處理快照：

如果儲存中有多個快照，範例選擇器會依版本順序處理可用的快照。
若快照出現順序錯亂（例如，snapshot_3 在 snapshot_4 之後到達），範例選擇器會排除該快照，因為其版本不高於上次提交的版本。
若無新的快照，版本函式或查詢則不會回傳結果，也不會進行處理。

步驟 1：準備快照檔案

建立包含快照資料的 CSV 檔案，並將其加入磁碟區或雲端儲存空間。檔案名稱依時間順序排列（例如， snapshot_1.csv， snapshot_2.csv）。

每個檔案都應包含userId和city的欄位。例如：

snapshot_1.csv：

userId	city
1	瓦哈卡州
2	蒙特雷
3	蒂華納

snapshot_2.csv：

userId	city
2	卡梅爾
3	洛杉磯
4	死亡谷

步驟 2：從快照執行 AUTO CDC 並使用版本函式

在管線中，在 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 Type 1，表格看起來和最近的快照完全一樣。差別在於下游查詢可以只處理變更記錄。

步驟 3：新增快照

在儲存位置新增一個 CSV 檔案，並包含修改後的資料（例如：更改城市值、增加新的資料列或刪除現有的資料列）。接著再跑一次管線來處理新的快照。

局限性

序列欄位必須是可排序的資料型態。 NULL 不支援排序值。
若要從 AUTO CDC 程序目標中串流資料，請從其變更摘要讀取。詳情請參閱「從 AUTO CDC 目標表讀取變更數據饋送」。

其他資源

變更資料擷取與快照：了解CDC的概念、快照及SCD類型。
透過以下方式AUTO CDC複製外部關聯式資料庫管理系統表格：學習如何使用once流程進行初始數據準備，然後繼續處理修改。
用管線回填歷史資料：學習如何回填僅追加和 AUTO CDC SCD 第一類目標表。
進階的自動CDC主題：了解針對自動CDC目標的變更操作、讀取變更資料流及處理指標。
回溯並重新執行管線：了解如何將 AUTO CDC 目標還原至較早的時間點，並重新處理變更。
教學課程：使用變更資料擷取建置 ETL 管線

意見反應

此頁面對您有幫助嗎？

上次更新時間： 2026-10-09