Habilite la captura de datos modificados para las conexiones de origen en la API

Utilice la captura de datos modificados en las fuentes de Adobe Experience Platform para mantener los sistemas de origen y destino sincronizados en tiempo casi real.

Experience Platform admite actualmente copia de datos incremental, que transfiere periódicamente registros recién creados o actualizados del sistema de origen a los conjuntos de datos ingeridos. Este método se basa en una columna de marca de tiempo para realizar el seguimiento de los cambios, pero no detecta las eliminaciones, lo que puede provocar incoherencias en los datos a lo largo del tiempo.

Por el contrario, change data capture captura y aplica inserciones, actualizaciones y eliminaciones en tiempo casi real. Este completo seguimiento de cambios garantiza que los conjuntos de datos permanezcan totalmente alineados con el sistema de origen y proporciona un historial de cambios completo, más allá de lo que admite la copia incremental. Sin embargo, las operaciones de eliminación requieren una consideración especial, ya que afectan a todas las aplicaciones que utilizan los conjuntos de datos de destinatario.

Para cambiar la captura de datos en Experience Platform se requiere Data Mirror con esquemas relacionales. Puede proporcionar datos de cambio a Data Mirror de dos formas:

Ambos enfoques requieren Data Mirror con esquemas relacionales para preservar las relaciones y exigir la exclusividad.

Data Mirror con esquemas relacionales

Data Mirror utiliza esquemas relacionales para ampliar la captura de datos modificados y habilitar funciones avanzadas de sincronización de bases de datos. Para obtener una descripción general de Data Mirror, consulte Información general de Data Mirror.

Los esquemas relacionales extienden Experience Platform para exigir la exclusividad de la clave principal, realizar un seguimiento de los cambios de nivel de fila y definir relaciones de nivel de esquema. Con la captura de datos modificados, aplican inserciones, actualizaciones y eliminaciones directamente en el lago de datos, lo que reduce la necesidad de Extraer, Transformar, Cargar (ETL) o reconciliación manual.

Consulte Información general sobre esquemas relacionales para obtener más información.

Requisitos de esquema relacional para capturar datos de cambio

Antes de utilizar un esquema relacional con captura de datos modificados, configure los siguientes identificadores:

Control de gestión de columnas control-column-handling

Utilice la columna _change_request_type para especificar cómo se debe procesar cada fila:

Esta columna solo se evalúa durante la ingesta y no se almacena ni asigna a campos XDM.

Flujo de trabajo workflow

Para habilitar la captura de datos modificados con un esquema relacional:

Proporcionar datos de cambio para orígenes basados en archivos file-based-sources

Para los orígenes basados en archivos (Amazon S3, Azure Blob, Google Cloud Storage y SFTP), incluya una columna _change_request_type en los archivos.

Use los valores de _change_request_type definidos en la sección Control column handling anterior.

Siga los pasos específicos de la fuente a continuación.

Fuentes de almacenamiento en nube cloud-storage-sources

Habilite la captura de datos modificados para las fuentes de almacenamiento en la nube siguiendo estos pasos:

Todos los orígenes de almacenamiento en la nube utilizan el mismo formato de columna _change_request_type descrito en la sección Orígenes basados en archivos anterior.

Ejemplo: habilitar la captura de datos de cambio al crear una conexión de origen de almacenamiento en la nube

Al crear su conexión de origen de almacenamiento en la nube, establezca params.cdcEnabled en true para habilitar la captura de datos modificados. Para los orígenes basados en archivos, las operaciones de cambio también dependen de la columna de control _change_request_type descrita anteriormente.

Formato de API

POST /sourceConnections

Solicitud

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"
    }
  }'

Respuesta

La respuesta devuelve el identificador único de la conexión de origen recién creada. Puede utilizar este ID en el siguiente paso del flujo de trabajo al crear un flujo de datos.

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

Para ver el flujo de trabajo completo del almacenamiento en la nube, incluido el esquema, el conjunto de datos, la conexión de destino, la asignación y la creación de flujo de datos, consulte Crear un flujo de datos para fuentes de almacenamiento en la nube mediante la Flow Service API.

Orígenes de base de datos database-sources

Azure Databricks

Para usar la captura de datos modificados con Azure Databricks, debe habilitar cambiar la fuente de datos en las tablas de origen y configurar Data Mirror con esquemas relacionales en Experience Platform.

Utilice los siguientes comandos para habilitar el cambio de fuente de datos en las tablas:

Nueva tabla

Para aplicar el cambio de fuente de datos a una nueva tabla, debe establecer la propiedad de tabla delta.enableChangeDataFeed en TRUE en el comando CREATE TABLE.

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

Tabla existente

Para aplicar el cambio de fuente de datos a una tabla existente, debe establecer la propiedad de tabla delta.enableChangeDataFeed en TRUE en el comando ALTER TABLE.

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

Todas las tablas nuevas

Para aplicar el cambio de fuente de datos a todas las tablas nuevas, debe establecer las propiedades predeterminadas en TRUE.

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

Para obtener más información, lea la Azure Databricks guía sobre cómo habilitar la fuente de datos para cambios.

Lea la siguiente documentación para ver los pasos que debe seguir para habilitar la captura de datos modificados para la conexión de origen de Azure Databricks:

Data Landing Zone

Para usar la captura de datos modificados con Data Landing Zone, debe habilitar cambiar la fuente de datos en las tablas de origen y configurar Data Mirror con esquemas relacionales en Experience Platform.

Lea la siguiente documentación para ver los pasos que debe seguir para habilitar la captura de datos modificados para la conexión de origen de Data Landing Zone:

Google BigQuery

Para usar la captura de datos modificados con Google BigQuery, debe habilitar el historial de cambios en las tablas de origen y configurar Data Mirror con esquemas relacionales en Experience Platform.

Para habilitar el historial de cambios en la conexión de origen de Google BigQuery, vaya a la página Google BigQuery en la consola Google Cloud y establezca enable_change_history en TRUE. Esta propiedad habilita el historial de cambios para la tabla de datos.

Para obtener más información, lea la guía de instrucciones de lenguaje de definición de datos en GoogleSQL.

Lea la siguiente documentación para ver los pasos que debe seguir para habilitar la captura de datos modificados para la conexión de origen de Google BigQuery:

Snowflake

Para usar la captura de datos modificados con Snowflake, debe habilitar el seguimiento de cambios en las tablas de origen y configurar Data Mirror con esquemas relacionales en Experience Platform.

En Snowflake, habilite el seguimiento de cambios usando ALTER TABLE y estableciendo CHANGE_TRACKING en TRUE.

ALTER TABLE mytable SET CHANGE_TRACKING = TRUE

Para obtener más información, lea la Snowflake guía sobre el uso de la cláusula de cambios.

Lea la siguiente documentación para ver los pasos que debe seguir para habilitar la captura de datos modificados para la conexión de origen de Snowflake:

Ejemplo: habilitar la captura de datos modificados al crear una conexión de origen de base de datos

Para los orígenes de base de datos admitidos, establezca params.cdcEnabled en true al crear la conexión de origen para habilitar la captura de datos modificados. Antes de utilizar esta opción, asegúrese de que el seguimiento de cambios está habilitado en el sistema de origen y que Data Mirror con esquemas relacionales está configurado en Experience Platform.

Este ejemplo se aplica a orígenes de base de datos admitidos que utilizan exportaciones de CDC nativas, como Azure Databricks, Google BigQuery y Snowflake.

Formato de API

POST /sourceConnections

Solicitud

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"
    }
  }'

Respuesta

La respuesta devuelve el identificador único de la conexión de origen recién creada. Puede utilizar este ID en el siguiente paso del flujo de trabajo al crear un flujo de datos.

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

Para ver el flujo de trabajo completo de ingesta de base de datos, incluida la conexión de destino, la asignación y la creación de flujo de datos, consulte Crear un flujo de datos para orígenes de base de datos mediante la Flow Service API.