Uso de Amazon Kinesis Data Streams como objetivo para AWS Database Migration Service

Puede utilizarlos AWS DMS para migrar datos a una transmisión de datos de Amazon Kinesis. Amazon Kinesis Data Streams forma parte del servicio de Amazon Kinesis Data Streams. Puede utilizar Kinesis Data Streams para recopilar y procesar grandes secuencias de registros de datos en tiempo real.

Un flujo de datos de Kinesis se compone de particiones. Las particiones son secuencias de registros de datos identificados inequívocamente en una secuencia. Para obtener más información sobre las particiones en Amazon Kinesis Data Streams, consulte Partición en la Guía para desarrolladores de Amazon Kinesis Data Streams.

AWS Database Migration Service publica registros en una transmisión de datos de Kinesis mediante JSON. Durante la conversión, AWS DMS serializa cada registro de la base de datos de origen en un par de atributo-valor en formato JSON o en un formato de mensaje JSON_UNFORMATTED. Un formato de mensaje JSON_UNFORMATTED es una cadena JSON de una sola línea con un delimitador de nueva línea. Permite a Amazon Data Firehose entregar los datos de Kinesis a un destino de Amazon S3 y, después, consultarlos mediante varios motores de consulta, incluido Amazon Athena.

Puede utilizar el mapeo de objetos para migrar sus datos desde cualquier origen de datos admitido a una secuencia de destino. Con el mapeo de datos, se determina cómo se estructuran los registros de datos de la secuencia. También debe definir una clave de partición para cada tabla, que Kinesis Data Streams utiliza para agrupar los datos en particiones.

Cuando AWS DMS crea tablas en un punto final de destino de Kinesis Data Streams, crea tantas tablas como en el punto final de la base de datos de origen. AWS DMS también establece varios valores de parámetros de Kinesis Data Streams. El costo de la creación de la tabla depende de la cantidad de datos y del número de tablas que hay que migrar.

nota

La opción de modo SSL de la AWS DMS consola o la API no se aplica a algunos servicios NoSQL y de streaming de datos, como Kinesis y DynamoDB. Son seguros de forma predeterminada, por lo que AWS DMS indica que la configuración del modo SSL es cero (Modo SSL=Ninguno). No necesita proporcionar ninguna configuración adicional para que el punto de conexión utilice SSL. Por ejemplo, cuando se utiliza Kinesis como punto de conexión de destino, es seguro de forma predeterminada. Todas las llamadas de API a Kinesis utilizan SSL, por lo que no es necesaria una opción SSL adicional en el AWS DMS punto final. Puede colocar y recuperar datos de forma segura a través de puntos de conexión SSL mediante el protocolo HTTPS, que AWS DMS utiliza de forma predeterminada al conectarse a un flujo de datos de Kinesis.

Configuración del punto de conexión de Kinesis Data Streams

Cuando utiliza los puntos de enlace de destino de Kinesis Data Streams, puede obtener los detalles de las transacciones y el control mediante KinesisSettings la opción de la API. AWS DMS

Puede configurar ajustes de conexión de las siguientes formas:

• En la AWS DMS consola, mediante la configuración de los puntos finales.

• En la CLI, mediante la kinesis-settings opción del CreateEndpointcomando.

En la CLI, utilice los siguientes parámetros de solicitud de la opción de kinesis-settings:

nota

La compatibilidad con la configuración del punto de conexión de IncludeNullAndEmpty está disponible en la versión de AWS DMS 3.4.1 y superiores. Sin embargo, la compatibilidad con las siguientes configuraciones de punto final para los destinos de Kinesis Data Streams está disponible en AWS DMS.

• MessageFormat: el formato del resultado de los registros creados en el punto de conexión. El formato del mensaje es JSON (predeterminado) o JSON_UNFORMATTED (una sola línea sin tabulación).

• IncludeControlDetails: muestra información detallada de control para la definición de tablas, la definición de columnas y los cambios de tablas y columnas en la salida del mensaje de Kinesis. El valor predeterminado es false.

• IncludeNullAndEmpty: incluya columnas NULL y vacías en el objetivo. El valor predeterminado es false.

• IncludePartitionValue: muestra el valor de partición dentro de la salida del mensaje de Kinesis, a menos que el tipo de partición sea schema-table-type. El valor predeterminado es false.

• IncludeTableAlterOperations: incluye todas las operaciones de lenguaje de definición de datos (DDL) que cambien la tabla en los datos de control, como rename-table, drop-table, add-column, drop-column y rename-column. El valor predeterminado es false.

• IncludeTransactionDetails: proporciona información detallada sobre transacciones de la base de datos de origen. Esta información incluye una marca temporal de confirmación, una posición de registro y valores para transaction_id, previous_transaction_id y transaction_record_id (el desplazamiento del registro dentro de una transacción). El valor predeterminado es false.

• PartitionIncludeSchemaTable: agrega los nombres de los esquemas y de las tablas como prefijo a los valores de partición, cuando el tipo de partición es primary-key-type. Al hacerlo, aumenta la distribución de datos entre las particiones de Kinesis. Por ejemplo, supongamos que un esquema SysBench tiene miles de tablas y cada tabla tiene un rango limitado para una clave principal. En este caso, la misma clave principal se envía desde miles de tablas a la misma partición, lo que provoca la limitación controlada. El valor predeterminado es false.

En el siguiente ejemplo, se muestra la opción kinesis-settings en uso con un comando create-endpoint de ejemplo emitido mediante la AWS CLI.

aws dms create-endpoint --endpoint-identifier=$target_name --engine-name kinesis --endpoint-type target
--region us-east-1 --kinesis-settings ServiceAccessRoleArn=arn:aws:iam::333333333333:role/dms-kinesis-role,
StreamArn=arn:aws:kinesis:us-east-1:333333333333:stream/dms-kinesis-target-doc,MessageFormat=json-unformatted,
IncludeControlDetails=true,IncludeTransactionDetails=true,IncludePartitionValue=true,PartitionIncludeSchemaTable=true,
IncludeTableAlterOperations=true

Configuración de tareas de carga completa con varios subprocesos

Para ayudar a aumentar la velocidad de la transferencia, AWS DMS admite una carga completa de subprocesos múltiples a una instancia de destino de Kinesis Data Streams. DMS admite este multriproceso con configuración de tareas que incluyen lo siguiente:

• MaxFullLoadSubTasks: utilice esta opción para indicar el número máximo de tablas de origen que se pueden cargar en paralelo. DMS carga cada tabla en su tabla de destino de Kinesis correspondiente mediante una tarea secundaria dedicada. El valor predeterminado es 8, el valor máximo es 49.

• ParallelLoadThreads— Utilice esta opción para especificar el número de subprocesos que se AWS DMS utilizan para cargar cada tabla en su tabla de destino de Kinesis. El valor máximo para un objetivo de Kinesis Data Streams es 32. Puede pedir que se incremente este límite máximo.

• ParallelLoadBufferSize: utilice esta opción para especificar el número máximo de registros para almacenar en el búfer que los subprocesos de carga en paralelo utilizan para cargar datos en el destino de Kinesis. El valor predeterminado es 50. El valor máximo es 1000. Utilice este parámetro con ParallelLoadThreads. ParallelLoadBufferSize es válido solo cuando hay más de un subproceso.

• ParallelLoadQueuesPerThread: utilice esta opción para especificar el número de colas que acceden a cada subproceso simultáneo para eliminar los registros de datos de las colas y generar una carga por lotes para el destino. El valor predeterminado de es 1. Sin embargo, para los destinos de Kinesis de varios tamaños de carga, el intervalo válido es de 5-512 colas por subproceso.

Configuración de tareas de carga de CDC con varios subprocesos

Puede mejorar el rendimiento de la captura de datos de cambios (CDC) para los puntos de conexión de destino de flujo de datos en tiempo real como Kinesis mediante la configuración de tareas para modificar el comportamiento de la llamada a la API PutRecords. Para ello, puede especificar el número de subprocesos simultáneos, las colas por subproceso y el número de registros que se van a almacenar en un búfer mediante la configuración de tareas ParallelApply*. Suponga, por ejemplo, que desea realizar una carga de CDC y aplicar 128 subprocesos en paralelo. También desea acceder a 64 colas por subproceso, con 50 registros almacenados por búfer.

Para promover el desempeño de los CDC, AWS DMS admite las siguientes configuraciones de tareas:

• ParallelApplyThreads— Especifica el número de subprocesos simultáneos que se AWS DMS utilizan durante una carga de CDC para enviar registros de datos a un punto final de Kinesis de destino. El valor predeterminado es cero (0) y el valor máximo es 32.

• ParallelApplyBufferSize: especifica el número máximo de registros que se almacenan en cada cola del búfer para los subprocesos simultáneos que insertan datos en un punto de conexión de destino de Kinesis durante una carga de CDC. El valor predeterminado es 100 y el máximo es 1000. Utilice esta opción cuando ParallelApplyThreads especifique más de un subproceso.

• ParallelApplyQueuesPerThread: especifica el número de colas a las que accede cada subproceso para sacar registros de datos de las colas y generar una carga por lotes para un punto de conexión de Kinesis durante el proceso de CDC. El valor predeterminado es 1 y el máximo es 512.

Cuando se utiliza la configuración de tareas ParallelApply*, el valor predeterminado de partition-key-type es el valor de primary-key de la tabla, no el valor de schema-name.table-name.

Uso de una imagen anterior para consultar los valores originales de las filas de CDC de un flujo de datos de Kinesis como destino

Al escribir actualizaciones de CDC en un destino de flujo de datos como Kinesis, puede ver los valores originales de una fila de base de datos de origen antes de cambiar mediante una actualización. Para que esto sea posible, AWS DMS rellena una imagen anterior de los eventos de actualización en función de los datos proporcionados por el motor de base de datos de origen.

Los diferentes motores de base de datos de origen proporcionan diferentes cantidades de información para una imagen anterior:

• Oracle proporciona actualizaciones a las columnas solo si cambian.

• PostgreSQL proporciona datos solo para las columnas que forman parte de la clave principal (cambiadas o no). Para proporcionar datos para todas las columnas (cambiadas o no), debe establecer REPLICA_IDENTITY en FULL en lugar de DEFAULT. Tenga en cuenta que debe elegir cuidadosamente la configuración de REPLICA_IDENTITY para cada tabla. Si establece REPLICA_IDENTITY en FULL, todos los valores de las columnas se escriben en el registro antes de la escritura (WAL) de forma continua. Esto puede provocar problemas de rendimiento o de recursos en las tablas que se actualizan con frecuencia.

• MySQL generalmente proporciona datos para todas las columnas excepto para los tipos de datos BLOB y CLOB (cambiadas o no).

Para habilitar imágenes anteriores para agregar valores originales de la base de datos de origen a la salida AWS DMS , utilice la configuración de tarea BeforeImageSettings o el parámetro add-before-image-columns. Este parámetro aplica una regla de transformación de columna.

BeforeImageSettings agrega un nuevo atributo JSON a cada operación de actualización con valores recopilados desde el sistema de base de datos de origen, como se muestra a continuación.

"BeforeImageSettings": {
    "EnableBeforeImage": boolean,
    "FieldName": string,  
    "ColumnFilter": pk-only (default) / non-lob / all (but only one)
}
        

nota

Solo se aplica BeforeImageSettings a AWS DMS las tareas que contienen un componente de los CDC, como la carga completa más las tareas de los CDC (que migran los datos existentes y reproducen los cambios en curso), o a las tareas exclusivas de los CDC (que solo replican los cambios en los datos). No se aplica BeforeImageSettings a tareas que son solo de carga completa.

Para las opciones de BeforeImageSettings, se aplica lo siguiente:

• Establezca la opción EnableBeforeImage para habilitar true antes de crear imágenes. El valor predeterminado es false.

• Utilice la opción FieldName para asignar un nombre al nuevo atributo JSON. Cuando EnableBeforeImage es true, FieldName es necesario y no puede estar vacío.

• La opción ColumnFilter especifica una columna para agregar mediante el uso de las imágenes anteriores. Para agregar solo columnas que forman parte de las claves principales de la tabla, utilice el valor predeterminado, pk-only. Para agregar cualquier columna que tenga un valor de imagen anterior, utilice all. Tenga en cuenta que la imagen anterior no contiene columnas con tipos de datos de LOB, como CLOB o BLOB.

  "BeforeImageSettings": {
      "EnableBeforeImage": true,
      "FieldName": "before-image",
      "ColumnFilter": "pk-only"
    }
  

nota

Los objetivos de Amazon S3 no admiten BeforeImageSettings. Para los destinos de S3, utilice solo la regla de transformación add-before-image-columns que debe realizarse antes de crear imágenes durante el CDC.

Uso de una regla de transformación de imagen anterior

Como alternativa a la configuración de tareas, puede utilizar el parámetro add-before-image-columns, que aplica una regla de transformación de columnas. Con este parámetro, puede habilitar imágenes anteriores durante CDC en destinos de flujo de datos como Kinesis.

Al utilizar add-before-image-columns en una regla de transformación, puede aplicar un control más detallado de los resultados de la imagen anterior. Las reglas de transformación permiten utilizar un localizador de objetos que le da control sobre las tablas seleccionadas para la regla. Además, puede encadenar reglas de transformación, lo que permite aplicar diferentes reglas a diferentes tablas. A continuación, puede manipular las columnas producidas utilizando otras reglas.

nota

No utilice el parámetro add-before-image-columns junto con la configuración de tarea BeforeImageSettings dentro de la misma tarea. En su lugar, utilice el parámetro o la configuración, pero no ambos, para una sola tarea.

Un tipo de regla transformation con el parámetro add-before-image-columns de una columna debe proporcionar una sección before-image-def. A continuación se muestra un ejemplo.

    {
      "rule-type": "transformation",
      …
      "rule-target": "column",
      "rule-action": "add-before-image-columns",
      "before-image-def":{
        "column-filter": one-of  (pk-only / non-lob / all),
        "column-prefix": string,
        "column-suffix": string,
      }
    }

El valor de column-prefix se antepone a un nombre de columna y el valor predeterminado de column-prefix es BI_. El valor de column-suffix se añade al nombre de la columna y el valor predeterminado está vacío. No configure ambas cadenas column-prefix y column-suffix como cadenas vacías.

Elija un valor para column-filter. Para agregar solo columnas que forman parte de las claves principales de la tabla, elija pk-only. Elija non-lob para agregar solo columnas que no sean de tipo LOB. O elija all para agregar cualquier columna que tenga un valor de imagen anterior.

Ejemplo de una regla de transformación de imagen anterior

La regla de transformación del siguiente ejemplo agrega una nueva columna llamada BI_emp_no en el destino. Entonces, una instrucción como UPDATE employees SET emp_no = 3 WHERE emp_no = 1; rellena el campo BI_emp_no con 1. Cuando escribe actualizaciones de CDC en destinos de Amazon S3, la columna BI_emp_no permite indicar qué fila original se actualizó.

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "1",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "rule-action": "include"
    },
    {
      "rule-type": "transformation",
      "rule-id": "2",
      "rule-name": "2",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "employees"
      },
      "rule-action": "add-before-image-columns",
      "before-image-def": {
        "column-prefix": "BI_",
        "column-suffix": "",
        "column-filter": "pk-only"
      }
    }
  ]
}

Para obtener información sobre el uso de la acción de regla add-before-image-columns, consulte Reglas y acciones de transformación.

Requisitos previos para utilizar una transmisión de datos de Kinesis como destino para AWS Database Migration Service

Función de IAM para utilizar una transmisión de datos de Kinesis como objetivo para AWS Database Migration Service

Antes de configurar una transmisión de datos de Kinesis como destino AWS DMS, asegúrese de crear una función de IAM. Esta función debe permitir asumir y conceder acceso AWS DMS a las transmisiones de datos de Kinesis a las que se están migrando. El conjunto mínimo de permisos de acceso se muestra en la siguiente política de IAM.

{
   "Version": "2012-10-17",
   "Statement": [
   {
     "Sid": "1",
     "Effect": "Allow",
     "Principal": {
        "Service": "dms.amazonaws.com"
     },
   "Action": "sts:AssumeRole"
   }
]
} 

        

El rol que utilice para la migración a un flujo de datos de Kinesis debe tener los siguientes permisos.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:region:accountID:stream/streamName"
        }
    ]
}

Acceder a una transmisión de datos de Kinesis como destino para AWS Database Migration Service

En AWS DMS la versión 3.4.7 y versiones posteriores, para conectarse a un endpoint de Kinesis, debe realizar una de las siguientes acciones:

• Configure DMS para que utilice puntos de conexión de VPC. Para obtener más información sobre la configuración de DMS para utilizar puntos de conexión de VPC, consulte Configuración de puntos de conexión de VPC como puntos de conexión de origen y destino de AWS DMS.

• Configure DMS para que utilice rutas públicas, es decir, haga pública su instancia de replicación. Para obtener información sobre las instancias de replicación públicas, consulte Instancias de replicación pública y privada.

Limitaciones al utilizar Kinesis Data Streams como objetivo para AWS Database Migration Service

Al utilizar Kinesis Data Streams como destino se aplican las siguientes restricciones:

• AWS DMS publica cada actualización en un único registro de la base de datos de origen como un registro de datos en un flujo de datos de Kinesis determinado, independientemente de las transacciones. Sin embargo, puede incluir detalles de transacción para cada registro de datos utilizando los parámetros pertinentes de la API KinesisSettings.

• No se admite el modo LOB completo.

• El tamaño de LOB máximo admitido es 1 MB.

• Kinesis Data Streams no admite la desduplicación. Las aplicaciones que consumen datos de una secuencia necesitan ocuparse de los registros duplicados. Para obtener más información, consulte Gestión de registros duplicados en la Guía para desarrolladores de Amazon Kinesis Data Streams.

• AWS DMS admite las dos formas siguientes de claves de partición:

  • SchemaName.TableName una combinación del nombre de esquema y de tabla.

  • ${AttributeName}: el valor de uno de los campos del archivo JSON o la clave principal de la tabla de la base de datos de origen.

• Para obtener información sobre el cifrado de los datos en reposo en Kinesis Data Streams, consulte Protección de datos en Kinesis Data Streams en la Guía para desarrolladores de AWS Key Management Service .

• BatchApply no es compatible con un punto de conexión de Kinesis. Es posible que el uso de la aplicación por lotes (por ejemplo, la configuración de tareas de metadatos de destino BatchApplyEnabled) para un objetivo de Kinesis provoque la pérdida de datos.

• Los destinos de Kinesis solo se admiten para una transmisión de datos de Kinesis en la misma AWS cuenta y en la Región de AWS misma instancia de replicación.

• Al migrar desde una fuente de MySQL, los BeforeImage datos no incluyen los tipos de datos CLOB y BLOB. Para obtener más información, consulte Uso de una imagen anterior para consultar los valores originales de las filas de CDC de un flujo de datos de Kinesis como destino.

• AWS DMS no admite la migración de valores de tipos de BigInt datos con más de 16 dígitos. Para evitar esta limitación, puede usar la siguiente regla de transformación para convertir la columna BigInt en una cadena. Para obtener más información sobre las reglas de transformación, consulte Reglas y acciones de transformación.

  {
      "rule-type": "transformation",
      "rule-id": "id",
      "rule-name": "name",
      "rule-target": "column",
      "object-locator": {
          "schema-name": "valid object-mapping rule action",
          "table-name": "",
          "column-name": ""
      },
      "rule-action": "change-data-type",
      "data-type": {
          "type": "string",
          "length": 20
      }
  }

Uso de la asignación de objetos para migrar datos a un flujo de datos de Kinesis

AWS DMS utiliza reglas de mapeo de tablas para mapear los datos del flujo de datos de Kinesis de origen al de destino. Para asignar datos a una secuencia de destino, se utiliza un tipo de regla de mapeo de tablas denominada "object mapping". Puede utilizar la asignación de objetos para definir cómo los registros de datos del origen asignan a los registros de datos publicados en el flujo de datos de Kinesis.

Kinesis Data Streams no tiene una estructura predeterminada distinta de una clave de partición. En una regla de asignación de objetos, los valores posibles de partition-key-type para los registros de datos son schema-table, transaction-id, primary-key, constant y attribute-name.

Para crear una regla de mapeo de objetos, especifique rule-type como object-mapping. Esta regla indica el tipo de mapeo de objetos que desea utilizar.

La estructura de la regla es la siguiente.

{
    "rules": [
        {
            "rule-type": "object-mapping",
            "rule-id": "id",
            "rule-name": "name",
            "rule-action": "valid object-mapping rule action",
            "object-locator": {
                "schema-name": "case-sensitive schema name",
                "table-name": ""
            }
        }
    ]
}

AWS DMS actualmente admite map-record-to-record y es map-record-to-document el único valor válido para el parámetro. rule-action Esta configuración afecta a los valores que no están excluidos como parte de la lista de atributos exclude-columns. Los map-record-to-document valores map-record-to-record y especifican cómo se AWS DMS gestionan estos registros de forma predeterminada. Estos valores no afectan a los mapeos de atributos en modo alguno.

Utilice map-record-to-record al migrar desde una base de datos relacional a un flujo de datos de Kinesis. Este tipo de regla utiliza el valor taskResourceId.schemaName.tableName de la base de datos relacional como la clave de partición en el flujo de datos de Kinesis y crea un atributo para cada columna de la base de datos de origen.

Cuando utilice map-record-to-record, tenga en cuenta lo siguiente:

• Esta configuración solo afecta a las columnas excluidas de la lista exclude-columns.

• Para cada columna de este tipo, AWS DMS crea un atributo correspondiente en el tema de destino.

• AWS DMS crea el atributo correspondiente independientemente de si la columna de origen se utiliza en una asignación de atributos.

Se utiliza map-record-to-document para colocar las columnas de origen en un único documento plano en la secuencia de destino correspondiente utilizando el nombre de atributo “_doc”. AWS DMS coloca los datos en un único mapa plano del origen denominado “_doc”. Esta colocación se aplica a cada columna de la tabla de origen que no se enumera en la lista de atributos exclude-columns.

Una forma de entender map-record-to-record es verlo en acción. En este ejemplo, imagine que empieza con una fila de una tabla de base de datos relacional con la estructura y los datos siguientes.

FirstName	LastName	StoreId	HomeAddress	HomePhone	WorkAddress	WorkPhone	DateofBirth
Randy	Marsh	5	221B Baker Street	1234567890	31 Spooner Street, Quahog	9876543210	02/29/1988

Para migrar esta información desde un esquema denominado Test a un flujo de datos de Kinesis, debe crear reglas para asignar los datos a la secuencia de destino. La siguiente regla ilustra la operación de asignación.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

A continuación, se ilustra el formato de registro resultante en el flujo de datos de Kinesis:

• StreamName: XXX

• PartitionKey: Test.Customers //schmaname.tableName

• Datos: //El siguiente mensaje JSON

  
    {
       "FirstName": "Randy",
       "LastName": "Marsh",
       "StoreId":  "5",
       "HomeAddress": "221B Baker Street",
       "HomePhone": "1234567890",
       "WorkAddress": "31 Spooner Street, Quahog",
       "WorkPhone": "9876543210",
       "DateOfBirth": "02/29/1988"
    }   
                  
            

Sin embargo, supongamos que utiliza las mismas reglas pero cambia el parámetro rule-action a map-record-to-document y excluye determinadas columnas. La siguiente regla ilustra la operación de asignación.

{
	"rules": [
	   {
			"rule-type": "selection",
			"rule-id": "1",
			"rule-name": "1",
			"rule-action": "include",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "%"
			}
		},
		{
			"rule-type": "object-mapping",
			"rule-id": "2",
			"rule-name": "DefaultMapToKinesis",
			"rule-action": "map-record-to-document",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "Customers"
			},
			"mapping-parameters": {
				"exclude-columns": [
					"homeaddress",
					"homephone",
					"workaddress",
					"workphone"
				]
			}
		}
	]
}                

En este caso, las columnas que no aparecen en el parámetro exclude-columns, FirstName, LastName, StoreId y DateOfBirth, se asignan a _doc. A continuación, se ilustra el formato de registro resultante.

       {
            "data":{
                "_doc":{
                    "FirstName": "Randy",
                    "LastName": "Marsh",
                    "StoreId":  "5",
                    "DateOfBirth": "02/29/1988"
                }
            }
        }
            

Reestructuración de datos con el mapeo de atributos

Puede reestructurar los datos mientras los migra a un flujo de datos de Kinesis mediante un mapa de atributos. Por ejemplo, es posible que desee combinar varios campos del origen en un único campo en el destino. El mapa de atributos siguiente ilustra cómo reestructurar los datos.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "rule-action": "include",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "2",
            "rule-name": "TransformToKinesis",
            "rule-action": "map-record-to-record",
            "target-table-name": "CustomerData",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            },
            "mapping-parameters": {
                "partition-key-type": "attribute-name",
                "partition-key-name": "CustomerName",
                "exclude-columns": [
                    "firstname",
                    "lastname",
                    "homeaddress",
                    "homephone",
                    "workaddress",
                    "workphone"
                ],
                "attribute-mappings": [
                    {
                        "target-attribute-name": "CustomerName",
                        "attribute-type": "scalar",
                        "attribute-sub-type": "string",
                        "value": "${lastname}, ${firstname}"
                    },
                    {
                        "target-attribute-name": "ContactDetails",
                        "attribute-type": "document",
                        "attribute-sub-type": "json",
                        "value": {
                            "Home": {
                                "Address": "${homeaddress}",
                                "Phone": "${homephone}"
                            },
                            "Work": {
                                "Address": "${workaddress}",
                                "Phone": "${workphone}"
                            }
                        }
                    }
                ]
            }
        }
    ]
}

Para establecer un valor constante para partition-key, especifique un valor de partition-key. Tal vez desee hacer esto, por ejemplo, para obligar a que todos los datos se almacenen en una única partición. El siguiente mapeo ilustra este enfoque.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "1",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "%"
            },
            "rule-action": "include"
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "1",
            "rule-name": "TransformToKinesis",
            "rule-action": "map-record-to-document",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer"
            },
            "mapping-parameters": {
                "partition-key": {
                    "value": "ConstantPartitionKey"
                },
                "exclude-columns": [
                    "FirstName",
                    "LastName",
                    "HomeAddress",
                    "HomePhone",
                    "WorkAddress",
                    "WorkPhone"
                ],
                "attribute-mappings": [
                    {
                        "attribute-name": "CustomerName",
                        "value": "${FirstName},${LastName}"
                    },
                    {
                        "attribute-name": "ContactDetails",
                        "value": {
                            "Home": {
                                "Address": "${HomeAddress}",
                                "Phone": "${HomePhone}"
                            },
                            "Work": {
                                "Address": "${WorkAddress}",
                                "Phone": "${WorkPhone}"
                            }
                        }
                    },
                    {
                        "attribute-name": "DateOfBirth",
                        "value": "${DateOfBirth}"
                    }
                ]
            }
        }
    ]
}

nota

El valor partition-key de un registro de control para una tabla específica es TaskId.SchemaName.TableName. El valor partition-key de un registro de control para una tabla específica es el TaskId de ese registro. La especificación de un valor partition-key en el mapeo de objetos no tiene ningún efecto en el elemento partition-key de un registro de control.

Formato de mensaje para Kinesis Data Streams

La salida JSON es simplemente una lista de pares de clave-valor. Un formato de mensaje JSON_UNFORMATTED es una cadena JSON de una sola línea con un delimitador de nueva línea.

AWS DMS proporciona los siguientes campos reservados para facilitar el consumo de los datos de Kinesis Data Streams:

RecordType

El tipo de registro puede ser de datos o de control. Los registros de datos representan las filas reales en el origen. Los registros de control son para eventos importantes de la secuencia como, por ejemplo, el reinicio de una tarea.

Operación

Para los registros de datos, la operación puede ser load, insert, update o delete.

Para los registros de control, la operación puede ser create-table, rename-table, drop-table, change-columns, add-column, drop-column, rename-column o column-type-change.

SchemaName

El esquema de origen del registro. Este campo puede estar vacío para un registro de control.

TableName

La tabla de origen del registro. Este campo puede estar vacío para un registro de control.

Timestamp

La marca temporal que indica cuándo se creó el mensaje JSON. El campo está formateado con el formato ISO 8601.