在API中啟用來源連線的變更資料擷取

在Adobe Experience Platform來源中使用變更資料擷取，讓您的來源和目的地系統近乎即時保持同步。

Experience Platform目前支援​增量資料副本，此副本會定期將來源系統新建立或更新後的記錄傳輸至擷取的資料集。 此方法依賴​ 時間戳記資料行 ​來追蹤變更，但不會偵測到刪除，而可能導致資料在一段時間內不一致。

相對地，變更資料擷取會擷取並近乎即時地套用插入、更新及刪除。 此完整的變更追蹤功能可確保資料集與來源系統保持完全一致，並提供完整的變更記錄，超出增量副本所支援的範圍。 不過，刪除作業需要特別考量，因為它們會影響使用目標資料集的所有應用程式。

Experience Platform中的變更資料擷取需要​ Data Mirror ​搭配關聯式結構描述。 您可以透過兩種方式將變更資料提供給Data Mirror：

兩種方法都需要使用關聯式結構的Data Mirror來保留關係並強制實施唯一性。

Data Mirror搭配關聯式結構描述

Data Mirror使用關聯式結構描述來擴充變更資料擷取及啟用進階資料庫同步功能。 如需Data Mirror的概觀，請參閱Data Mirror概觀。

關聯式結構描述可擴充Experience Platform以強制執行主鍵唯一性、追蹤列層級變更及定義結構描述層級關係。 透過變更資料擷取，它們直接在資料湖中套用插入、更新及刪除，減少擷取、轉換、載入(ETL)或手動調解的需求。

如需詳細資訊，請參閱關聯式結構描述概觀。

變更資料擷取的關聯式結構描述需求

在將關聯式綱要用於變更資料擷取之前，請設定下列識別碼：

控制欄處理 control-column-handling

使用_change_request_type欄指定每個資料列的處理方式：

此欄僅會在擷取期間評估，不會儲存或對應至XDM欄位。

工作流程 workflow

若要啟用關聯式綱要的變更資料擷取：

提供檔案型來源的變更資料 file-based-sources

針對檔案型來源（Amazon S3、Azure Blob、Google Cloud Storage和SFTP），請在您的檔案中包含_change_request_type欄。

使用上面控制項資料行處理區段中定義的_change_request_type值。

請遵循下列來源專屬步驟。

雲端儲存空間來源 cloud-storage-sources

請依照下列步驟啟用雲端儲存空間來源的變更資料擷取：

所有雲端儲存空間來源都使用上述檔案型來源一節中所述的相同_change_request_type欄格式。

範例：建立雲端儲存來源連線時啟用變更資料擷取

建立您的雲端儲存空間來源連線時，請將params.cdcEnabled設定為true以啟用變更資料擷取。 對於以檔案為基礎的來源，變更作業也需仰賴上述的_change_request_type控制項資料行。

API格式

POST /sourceConnections

要求

curl -X POST \
  'https://platform.adobe.io/data/foundation/flowservice/sourceConnections' \
  -H 'Authorization: Bearer {ACCESS_TOKEN}' \
  -H 'x-api-key: {API_KEY}' \
  -H 'x-gw-ims-org-id: {ORG_ID}' \
  -H 'x-sandbox-name: {SANDBOX_NAME}' \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "Cloud Storage source connection with CDC enabled",
    "description": "Source connection for ingesting change data from cloud storage",
    "baseConnectionId": "{BASE_CONNECTION_ID}",
    "data": {
      "format": "delimited"
    },
    "params": {
      "path": "/acme/cdc/account.csv",
      "type": "file",
      "cdcEnabled": true
    },
    "connectionSpec": {
      "id": "{CONNECTION_SPEC_ID}",
      "version": "1.0"
    }
  }'

回應

回應會傳回新建立的來源連線的唯一識別碼。 建立資料流時，您可以在工作流程的下一個步驟中使用此ID。

{
  "id": "26b53912-1005-49f0-b539-12100559f0e2",
  "etag": "\"11004d97-0000-0200-0000-5f3c3b140000\""
}

如需完整的雲端儲存工作流程，包括結構描述、資料集、目標連線、對應和資料流建立，請參閱使用 Flow Service API建立雲端儲存來源的資料流。

資料庫來源 database-sources

Azure Databricks

若要搭配Azure Databricks使用變更資料擷取，您必須在來源資料表中啟用​變更資料摘要，並在Experience Platform中透過relational schema設定Data Mirror。

使用下列命令來啟用表格上的變更資料摘要：

新資料表

若要將變更資料摘要套用至新的資料表，您必須在CREATE TABLE命令中將資料表屬性delta.enableChangeDataFeed設定為TRUE。

CREATE TABLE student (id INT, name STRING, age INT) TBLPROPERTIES (delta.enableChangeDataFeed = true)

現有的資料表

若要將變更資料摘要套用至現有的資料表，您必須在ALTER TABLE命令中將資料表屬性delta.enableChangeDataFeed設定為TRUE。

ALTER TABLE myDeltaTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

所有新資料表

若要將變更資料摘要套用至所有新表格，您必須將預設屬性設定為TRUE。

set spark.databricks.delta.properties.defaults.enableChangeDataFeed = true;

如需詳細資訊，請閱讀啟用變更資料摘要🔗的Azure Databricks 指南。

請閱讀下列檔案，以瞭解如何為Azure Databricks來源連線啟用變更資料擷取的步驟：

Data Landing Zone

若要搭配Data Landing Zone使用變更資料擷取，您必須在來源資料表中啟用​變更資料摘要，並在Experience Platform中透過relational schema設定Data Mirror。

請閱讀下列檔案，以瞭解如何為Data Landing Zone來源連線啟用變更資料擷取的步驟：

Google BigQuery

若要搭配Google BigQuery使用變更資料擷取，您必須在來源表格中啟用變更歷史記錄，並在Experience Platform中以relational schema設定Data Mirror。

若要在您的Google BigQuery來源連線中啟用變更記錄，請瀏覽至Google Cloud主控台中的Google BigQuery頁面，並將enable_change_history設定為TRUE。 此屬性可啟用資料表變更記錄。

如需詳細資訊，請閱讀 GoogleSQL🔗中資料定義語言陳述式的指南。

請閱讀下列檔案，以瞭解如何為Google BigQuery來源連線啟用變更資料擷取的步驟：

Snowflake

若要搭配Snowflake使用變更資料擷取，您必須在來源資料表中啟用​變更追蹤，並在Experience Platform中透過relational schema設定Data Mirror。

在Snowflake中，使用ALTER TABLE並設定CHANGE_TRACKING為TRUE來啟用變更追蹤。

ALTER TABLE mytable SET CHANGE_TRACKING = TRUE

如需詳細資訊，請閱讀使用changes子句🔗的Snowflake 指南。

請閱讀下列檔案，以瞭解如何為Snowflake來源連線啟用變更資料擷取的步驟：

範例：建立資料庫來源連線時啟用變更資料擷取

針對支援的資料庫來源，在建立來源連線以啟用變更資料擷取時，請將params.cdcEnabled設為true。 使用此選項之前，請確定您的來源系統中已啟用變更追蹤，並且已在Experience Platform中設定具有關聯式結構描述的Data Mirror 。

此範例適用於使用原生CDC匯出的支援資料庫來源，例如Azure Databricks、Google BigQuery和Snowflake。

API格式

POST /sourceConnections

要求

curl -X POST \
  'https://platform.adobe.io/data/foundation/flowservice/sourceConnections' \
  -H 'Authorization: Bearer {ACCESS_TOKEN}' \
  -H 'x-api-key: {API_KEY}' \
  -H 'x-gw-ims-org-id: {ORG_ID}' \
  -H 'x-sandbox-name: {SANDBOX_NAME}' \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "Database source connection with CDC enabled",
    "description": "Source connection for ingesting change data from a database source",
    "baseConnectionId": "{BASE_CONNECTION_ID}",
    "data": {
      "format": "tabular"
    },
    "params": {
      "tableName": "{TABLE_NAME}",
      "columns": [
        {
          "name": "id",
          "type": "string",
          "xdm": {
            "type": "string"
          }
        },
        {
          "name": "name",
          "type": "string",
          "xdm": {
            "type": "string"
          }
        },
        {
          "name": "lastModified",
          "type": "string",
          "meta:xdmType": "date-time",
          "xdm": {
            "type": "string",
            "format": "date-time"
          }
        }
      ],
      "cdcEnabled": true
    },
    "connectionSpec": {
      "id": "{CONNECTION_SPEC_ID}",
      "version": "1.0"
    }
  }'

回應

回應會傳回新建立的來源連線的唯一識別碼。 建立資料流時，您可以在工作流程的下一個步驟中使用此ID。

{
  "id": "b7581b59-c603-4df1-a689-d23d7ac440f3",
  "etag": "\"ef05d265-0000-0200-0000-6019e0080000\""
}

如需完整的資料庫擷取工作流程，包括目標連線、對應和資料流建立，請參閱使用 Flow Service API建立資料庫來源的資料流。