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

Puede utilizar AWS DMS para migrar datos a Amazon Kinesis Data Streams. 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 un flujo 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 enviar datos de Kinesis a un destino de Amazon S3 y consultar después estos datos mediante distintos 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 de conexión de destino de Kinesis Data Streams, crea tantas tablas como haya en el punto de conexión 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.

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

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 basada en 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:

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)
}
        

Para las opciones de BeforeImageSettings, se aplica lo siguiente:

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.

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 el uso de un flujo de datos de Kinesis como destino para AWS Database Migration Service

Rol de IAM para el uso de un flujo de datos de Kinesis como destino para AWS Database Migration Service

Antes de configurar un flujo de datos de Kinesis como destino para AWS DMS, asegúrese de crear un rol de IAM. Este rol debe permitir a AWS DMS asumir y conceder acceso a Kinesis Data Streams que se va a migrar. 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 un flujo de datos de Kinesis como destino para AWS Database Migration Service

En AWS DMS versión 3.4.7 y posteriores, para conectarse a un punto de conexión de Kinesis, debe realizar una de estas acciones:

Restricciones al usar Kinesis Data Streams como destino para AWS Database Migration Service

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

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

AWS DMS usa reglas de asignación de tablas para asignar datos del origen al flujo de datos de Kinesis 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 map-record-to-document como únicos valores válidos 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 valores map-record-to-record y map-record-to-document especifican cómo AWS DMS gestiona 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:

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.

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:

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

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 datos desde Kinesis Data Streams: