在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中用“关系架构”配置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为数据库源创建数据流。