Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Uso de Apache Kafka como objetivo para AWS Database Migration Service

Puede usarlo AWS DMS para migrar datos a un clúster de Apache Kafka. Apache Kafka es una plataforma de streaming distribuida. Puede utilizar Apache Kafka para ingerir y procesar datos de streaming en tiempo real.

AWS también ofrece Amazon Managed Streaming para que Apache Kafka MSK (Amazon) lo utilice como objetivo. AWS DMS Amazon MSK es un servicio de streaming de Apache Kafka totalmente gestionado que simplifica la implementación y la gestión de las instancias de Apache Kafka. Funciona con versiones de código abierto de Apache Kafka y se accede a las MSK instancias de Amazon como AWS DMS destinos exactamente igual que a cualquier instancia de Apache Kafka. Para obtener más información, consulta ¿Qué es AmazonMSK? en la Guía para desarrolladores de Amazon Managed Streaming for Apache Kafka.

Un clúster de Kafka almacena flujos de registros en categorías denominadas temas que se dividen en particiones. Las particiones son secuencias de registros de datos (mensajes) identificados de forma única en un tema. Las particiones se pueden distribuir entre varios agentes de un clúster para permitir el procesamiento paralelo de los registros de un tema. Para obtener más información sobre temas y particiones y su distribución en Apache Kafka, consulte Temas y registros y Distribución.

El clúster de Kafka puede ser una MSK instancia de Amazon, un clúster que se ejecute en una EC2 instancia de Amazon o un clúster local. Una MSK instancia de Amazon o un clúster de una EC2 instancia de Amazon pueden estar en la misma instancia VPC o en una diferente. Si el clúster está en las instalaciones, puede usar su propio servidor de nombres en las instalaciones para la instancia de replicación a fin de resolver el nombre de host del clúster. Para obtener información acerca de cómo configurar un servidor de nombres para la instancia de replicación, consulte Uso de su propio servidor de nombres en las instalaciones. Para obtener más información sobre cómo configurar una red, consulte Configuración de una red para una instancia de replicación.

Cuando utilice un MSK clúster de Amazon, asegúrese de que su grupo de seguridad permita el acceso desde la instancia de replicación. Para obtener información sobre cómo cambiar el grupo de seguridad de un MSK clúster de Amazon, consulte Cambiar el grupo de seguridad de un MSK clúster de Amazon.

AWS Database Migration Service publica registros sobre un tema de Kafka utilizandoJSON. Durante la conversión, AWS DMS serializa cada registro de la base de datos de origen en un par atributo-valor en formato. JSON

Para migrar los datos desde cualquier origen de datos admitido a un clúster de Kafka de destino, se usa la asignación de objetos. Con la asignación de objetos, se determina cómo se estructuran los registros de datos en el tema de destino. También debe definir una clave de partición para cada tabla, que Apache Kafka utiliza para agrupar los datos en particiones.

Actualmente, AWS DMS admite un solo tema por tarea. En el caso de una sola tarea con varias tablas, todos los mensajes van a un solo tema. Cada mensaje incluye una sección de metadatos que identifica el esquema y la tabla de destino. AWS DMS las versiones 3.4.6 y superiores admiten la replicación multitema mediante el mapeo de objetos. Para obtener más información, consulte Replicación multitemática mediante asignación de objetos.

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.

Conexión a Kafka mediante Transport Layer Security () TLS

Un clúster de Kafka acepta conexiones seguras mediante Transport Layer Security ()TLS. ConDMS, puede utilizar cualquiera de las siguientes tres opciones de protocolos de seguridad para proteger una conexión de punto final de Kafka.

Uso del SSL cifrado con Amazon MSK o un clúster Kafka autogestionado

Puedes usar el SSL cifrado para proteger una conexión de punto final a Amazon MSK o a un clúster de Kafka autogestionado. Al utilizar el método de autenticación SSL cifrada, los clientes validan la identidad de un servidor mediante el certificado del servidor. A continuación, se establece una conexión cifrada entre el servidor y el cliente.

"KafkaSettings": {
    "SecurityProtocol": "ssl-encryption", 
}
                

Uso de la autenticación de SSL

Puedes usar la SSL autenticación para proteger una conexión de punto final a Amazon MSK o a un clúster de Kafka autogestionado.

Para habilitar la autenticación y el cifrado de clientes mediante la SSL autenticación para conectarse a AmazonMSK, haga lo siguiente:

Uso deSASL: SSL autenticación para conectarse a Amazon MSK

El método de capa de seguridad y autenticación simple (SASL) utiliza un nombre de usuario y una contraseña para validar la identidad del cliente y establece una conexión cifrada entre el servidor y el cliente.

Para usarloSASL, primero debes crear un nombre de usuario y una contraseña seguros al configurar tu MSK clúster de Amazon. Para obtener una descripción de cómo configurar un nombre de usuario y una contraseña seguros para un MSK clúster de Amazon, consulte Configuración SASL o SCRAM autenticación para un MSK clúster de Amazon en la Guía para desarrolladores de Amazon Managed Streaming for Apache Kafka.

A continuación, cuando cree el punto de conexión de destino de Kafka, establezca la configuración del punto de conexión del protocolo de seguridad (SecurityProtocol) mediante la opción sasl-ssl. Establezca también las opciones SaslUsername y SaslPassword. Asegúrate de que sean coherentes con el nombre de usuario y la contraseña seguros que creaste al configurar tu MSK clúster de Amazon por primera vez, como se muestra en el siguiente JSON ejemplo.

                   
"KafkaSettings": {
    "SecurityProtocol": "sasl-ssl",
    "SaslUsername":"Amazon MSK cluster secure user name",
    "SaslPassword":"Amazon MSK cluster secure password"                    
}
                    

Utilice una imagen anterior para ver los valores originales de CDC las filas de Apache Kafka como objetivo

Al escribir CDC actualizaciones en un destino de transmisión de datos como Kafka, puede ver los valores originales de la fila de la base de datos de origen antes de cambiarlos 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:

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.

BeforeImageSettingsagrega un nuevo JSON atributo a cada operación de actualización con valores recopilados del 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 activarlo antes de crear imágenes CDC en destinos de transmisión de datos como Kafka.

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 añadir solo columnas que no sean de LOB tipo. 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 CDC actualizaciones en los destinos de Amazon S3, la BI_emp_no columna permite saber 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.

Limitaciones al utilizar Apache Kafka como objetivo para AWS Database Migration Service

Al utilizar Apache Kafka como destino, se aplican las siguientes restricciones:

Uso de la asignación de objetos para migrar datos a un tema de Kafka

AWS DMS utiliza reglas de mapeo de tablas para mapear datos del tema de Kafka de origen al de destino. Para asignar datos a un tema de destino, se utiliza un tipo de regla de asignación de tablas denominado asignación de objetos. Puede utilizar el mapeo de objetos para definir cómo los registros de datos del origen se asignan a los registros de datos publicados en un tema de Kafka.

Los temas de Kafka no tienen una estructura predeterminada distinta de una clave de partición.

Para crear una regla de mapeo de objetos, se especifica 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 rule-action parámetro. 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 tema de Kafka. Este tipo de regla utiliza el valor taskResourceId.schemaName.tableName de la base de datos relacional como la clave de partición en el tema de Kafka 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:

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 tema de Kafka, cree reglas para mapear los datos al tema. 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": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

Dados un tema de Kafka y una clave de partición (en este caso, taskResourceId.schemaName.tableName), a continuación se ilustra el formato de registro resultante al usar nuestros datos de ejemplo en el tema de destino de Kafka:

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

Reestructuración de datos con el mapeo de atributos

Puede reestructurar los datos mientras los migra a un tema de Kafka utilizando 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": "TransformToKafka",
            "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": "TransformToKafka",
            "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}"
                    }
                ]
            }
        }
    ]
}

Replicación multitemática mediante asignación de objetos

De forma predeterminada, AWS DMS las tareas migran todos los datos de origen a uno de los siguientes temas de Kafka:

Con las versiones 3.4.6 y posteriores AWS DMS del motor, puede usar el kafka-target-topic atributo para asignar cada tabla fuente migrada a un tema diferente. Por ejemplo, las siguientes reglas de asignación de objetos migran las tablas de origen Customer y Address a los temas de Kafka customer_topic y address_topic, respectivamente. Al mismo tiempo, AWS DMS migra todas las demás tablas de origen, incluida la Bills tabla del Test esquema, al tema especificado en el punto final de destino.

{
    "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": "MapToKafka1",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "customer_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customer" 
            },
            "partition-key": {"value": "ConstantPartitionKey" }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "3",
            "rule-name": "MapToKafka2",
            "rule-action": "map-record-to-record",
            "kafka-target-topic": "address_topic",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Address"
            },
            "partition-key": {"value": "HomeAddress" }
        },
        {
            "rule-type": "object-mapping",
            "rule-id": "4",
            "rule-name": "DefaultMapToKafka",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Bills"
            }
        }
    ]
}

Al utilizar la replicación multitema de Kafka, puede agrupar y migrar las tablas de origen para separar los temas de Kafka mediante una única tarea de replicación.

Formato de mensajes para Apache Kafka

El JSON resultado es simplemente una lista de pares clave-valor.

El siguiente ejemplo de JSON mensaje ilustra un mensaje de tipo de datos con todos los metadatos adicionales.

{ 
   "data":{ 
      "id":100000161,
      "fname":"val61s",
      "lname":"val61s",
      "REGION":"val61s"
   },
   "metadata":{ 
      "timestamp":"2019-10-31T22:53:59.721201Z",
      "record-type":"data",
      "operation":"insert",
      "partition-key-type":"primary-key",
      "partition-key-value":"sbtest.sbtest_x.100000161",
      "schema-name":"sbtest",
      "table-name":"sbtest_x",
      "transaction-id":9324410911751,
      "transaction-record-id":1,
      "prev-transaction-id":9324410910341,
      "prev-transaction-record-id":10,
      "commit-timestamp":"2019-10-31T22:53:55.000000Z",
      "stream-position":"mysql-bin-changelog.002171:36912271:0:36912333:9324410911751:mysql-bin-changelog.002171:36912209"
   }
}
                

El siguiente ejemplo de JSON mensaje ilustra un mensaje de tipo control.

{ 
   "control":{ 
      "table-def":{ 
         "columns":{ 
            "id":{ 
               "type":"WSTRING",
               "length":512,
               "nullable":false
            },
            "fname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "lname":{ 
               "type":"WSTRING",
               "length":255,
               "nullable":true
            },
            "REGION":{ 
               "type":"WSTRING",
               "length":1000,
               "nullable":true
            }
         },
         "primary-key":[ 
            "id"
         ],
         "collation-name":"latin1_swedish_ci"
      }
   },
   "metadata":{ 
      "timestamp":"2019-11-21T19:14:22.223792Z",
      "record-type":"control",
      "operation":"create-table",
      "partition-key-type":"task-id",
      "schema-name":"sbtest",
      "table-name":"sbtest_t1"
   }
}