Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Utiliser Apache Kafka comme cible pour AWS Database Migration Service

Vous pouvez l'utiliser AWS DMS pour migrer des données vers un cluster Apache Kafka. Apache Kafka est une plateforme de streaming distribuée. Apache Kafka vous permet d’ingérer et de traiter des données de streaming en temps réel.

AWS propose également Amazon Managed Streaming pour Apache Kafka (Amazon MSK) à utiliser comme cible. AWS DMS Amazon MSK est un service de streaming Apache Kafka entièrement géré qui simplifie l’implémentation et la gestion des instances Apache Kafka. Il fonctionne avec les versions open source d'Apache Kafka, et vous accédez aux instances Amazon MSK en tant que AWS DMS cibles, exactement comme n'importe quelle instance Apache Kafka. Pour plus d’informations, consultez Qu’est-ce qu’Amazon MSK ? dans le Guide du développeur Amazon Managed Streaming for Apache Kafka.

Un cluster Kafka stocke les flux d’enregistrements dans des catégories appelées « rubriques », divisées en partitions. Les partitions sont des séquences identifiées de manière unique d’enregistrements de données (messages) dans une rubrique. Les partitions peuvent être réparties entre plusieurs agents dans un cluster pour permettre le traitement parallèle des enregistrements d’une rubrique. Pour de plus amples informations sur les rubriques et les partitions et leur distribution dans Apache Kafka, veuillez consulter Rubriques et journaux et distribution.

Votre cluster Kafka peut être une instance Amazon MSK, un cluster exécuté sur une instance Amazon EC2 ou un cluster sur site. Une instance Amazon MSK ou un cluster sur une instance Amazon EC2 peut se trouver dans le même VPC ou dans un autre VPC. Dans le cas d’un cluster sur site, vous pouvez utiliser votre propre serveur de noms sur site pour votre instance de réplication afin de résoudre le nom d’hôte du cluster. Pour en savoir plus sur la configuration d’un serveur de noms pour votre instance de réplication, consultez Utilisation de votre propre serveur de noms sur site. Pour plus d’informations sur la configuration d’un réseau, consultez Configuration d'un réseau pour une instance de réplication.

Lorsque vous utilisez un cluster Amazon MSK, assurez-vous que son groupe de sécurité autorise l’accès à partir de votre instance de réplication. Pour en savoir plus sur la modification du groupe de sécurité d’un cluster Amazon MSK, consultez Modification du groupe de sécurité d’un cluster Amazon MSK.

AWS Database Migration Service publie des enregistrements dans un sujet Kafka à l'aide de JSON. Au cours de la conversion, AWS DMS sérialise chaque enregistrement de la base de données source dans une paire attribut-valeur au format JSON.

Utilisez le mappage d’objet pour migrer vos données de n’importe quelle source de données prise en charge vers un cluster Kafka cible. Avec le mappage d’objet, vous déterminez la façon de structurer les enregistrements de données dans la rubrique cible. Vous définissez également une clé de partition pour chaque table, qu'Apache Kafka utilise pour regrouper les données dans ses partitions.

Actuellement, ne AWS DMS prend en charge qu'un seul sujet par tâche. Pour une seule tâche comportant plusieurs tables, tous les messages sont placés dans une seule rubrique. Chaque message inclut une section de métadonnées qui identifie le schéma et la table cibles. AWS DMS les versions 3.4.6 et supérieures prennent en charge la réplication multisujet à l'aide du mappage d'objets. Pour de plus amples informations, veuillez consulter Réplication à plusieurs rubriques à l’aide du mappage d’objet.

Lorsque vous utilisez les paramètres de tâche ParallelApply*, la valeur par défaut partition-key-type est la valeur primary-key de la table, pas schema-name.table-name.

Connexion à Kafka à l’aide du protocole TLS (Transport Layer Security)

Un cluster Kafka accepte les connexions sécurisées à l’aide du protocole TLS (Transport Layer Security). Avec DMS, vous pouvez utiliser l’une des trois options de protocole de sécurité suivantes pour sécuriser la connexion d’un point de terminaison Kafka.

Utilisation du chiffrement SSL avec Amazon MSK ou un cluster Kafka autogéré

Vous pouvez utiliser le chiffrement SSL pour sécuriser la connexion d’un point de terminaison à Amazon MSK ou à un cluster Kafka autogéré. Lorsque vous utilisez la méthode d’authentification par chiffrement SSL, les clients valident l’identité d’un serveur par le biais du certificat du serveur. Une connexion chiffrée est alors établie entre le serveur et le client.

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

Utilisation de l’authentification SSL

Vous pouvez utiliser l’authentification SSL pour sécuriser la connexion d’un point de terminaison à Amazon MSK ou à un cluster Kafka autogéré.

Pour activer l’authentification client et le chiffrement à l’aide de l’authentification SSL pour se connecter à Amazon MSK, procédez comme suit :

Utilisation de l’authentification SASL-SSL pour se connecter à Amazon MSK

La méthode SASL (Simple Authentication and Security Layer) utilise un nom d’utilisateur et un mot de passe pour valider l’identité d’un client et établit une connexion chiffrée entre le serveur et le client.

Pour utiliser SASL, vous devez commencer par créer un nom d’utilisateur et un mot de passe sécurisés lorsque vous configurez votre cluster Amazon MSK. Pour savoir comment configurer un nom d’utilisateur et un mot de passe sécurisés pour un cluster Amazon MSK, consultez Configuration de l’authentification SASL/SCRAM pour un cluster Amazon MSK dans le Guide du développeur Amazon Managed Streaming for Apache Kafka.

Ensuite, lorsque vous créez votre point de terminaison cible Kafka, définissez le paramètre de point de terminaison du protocole de sécurité (SecurityProtocol) à l’aide de l’option sasl-ssl. Vous définissez également les options SaslUsername et SaslPassword. Assurez-vous qu’elles sont cohérentes avec le nom d’utilisateur et le mot de passe sécurisés que vous avez créés lorsque vous avez configuré votre cluster Amazon MSK pour la première fois, comme illustré dans l’exemple JSON suivant.

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

Utilisation d'une image antérieure pour afficher les valeurs originales des lignes de la CDC pour Apache Kafka comme cible

Lorsque vous écrivez des mises à jour de CDC sur une cible de diffusion de données comme Kafka, vous pouvez afficher les valeurs d'origine d'une ligne de base de données source avant de les modifier par une mise à jour. Pour ce faire, AWS DMS remplit une image antérieure des événements de mise à jour en fonction des données fournies par le moteur de base de données source.

Différents moteurs de base de données source fournissent différentes quantités d'informations pour une image antérieure :

Pour activer avant l'imagerie pour ajouter des valeurs d'origine de la base de données source à la sortie AWS DMS , utilisez le paramètre de tâche BeforeImageSettings ou le paramètre add-before-image-columns. Ce paramètre applique une règle de transformation de colonne.

BeforeImageSettings ajoute un nouvel attribut JSON à chaque opération de mise à jour avec des valeurs collectées à partir du système de base de données source, comme indiqué ci-dessous.

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

Pour les options BeforeImageSettings, les conditions suivantes s'appliquent :

Utilisation d'une règle de transformation d'image antérieure

Au lieu des paramètres de tâche, vous pouvez utiliser le paramètre add-before-image-columns, qui applique une règle de transformation de colonne. Avec ce paramètre, vous pouvez activer la génération d’image antérieure pendant la CDC sur des cibles de diffusion de données telles que Kafka.

En utilisant add-before-image-columns dans une règle de transformation, vous pouvez exercer un contrôle plus précis sur les résultats de l'image antérieure. Les règles de transformation vous permettent d'utiliser un localisateur d'objets qui vous fournit un contrôle sur les tables sélectionnées pour la règle. En outre, vous pouvez enchaîner les règles de transformation, ce qui permet d'appliquer différentes règles à différentes tables. Vous pouvez ensuite manipuler les colonnes produites à l'aide d'autres règles.

Un type de règle transformation avec le paramètre add-before-image-columns d'une colonne doit fournir une section before-image-def. Vous en trouverez un exemple ci-dessous.

    {
      "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,
      }
    }

La valeur de column-prefix est ajoutée à un nom de colonne et la valeur par défaut de column-prefix est BI_. La valeur de column-suffix est ajoutée au nom de la colonne et la valeur par défaut est vide. Ne définissez pas les deux column-prefix et column-suffix sur des chaînes vides.

Choisissez une valeur pour column-filter. Pour ajouter uniquement les colonnes qui font partie des clés primaires de la table, choisissez pk-only . Choisissez non-lob d'ajouter uniquement des colonnes qui ne sont pas de type LOB. Ou choisissez all d'ajouter une colonne qui a une valeur d’image antérieure.

Exemple de règle de transformation d'image antérieure

La règle de transformation de l'exemple suivant ajoute une nouvelle colonne appelée BI_emp_no dans la cible. Ainsi, une instruction comme UPDATE employees SET emp_no = 3 WHERE emp_no = 1; remplit le champ BI_emp_no avec 1. Lorsque vous écrivez des mises à jour de CDC sur des cibles Amazon S3, la colonne BI_emp_no permet de savoir quelle ligne d’origine a été mise à jour.

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

Pour plus d'informations sur l'utilisation de l'action de règle add-before-image-columns, consultez Règles et actions de transformation.

Limitations liées à l'utilisation d'Apache Kafka comme cible pour AWS Database Migration Service

Les limitations suivantes s'appliquent lorsque vous utilisez apache Kafka comme cible :

Utilisation du mappage d’objet pour migrer les données vers une rubrique Kafka

AWS DMS utilise des règles de mappage de tables pour mapper les données de la source vers le sujet Kafka cible. Pour mapper des données à une rubrique cible, vous utilisez un type de règle de mappage de table appelé « mappage d’objet ». Vous utilisez le mappage d'objet pour définir la façon dont les enregistrements de données de la source sont mappés sur les enregistrements de données publiés dans une rubrique Kafka.

Les rubriques Kafka ne disposent pas d'une structure prédéfinie autre que le fait d'avoir une clé de partition.

Pour créer une règle de mappage d'objet, spécifiez rule-type comme object-mapping. Cette règle spécifie le type de mappage d'objet que vous souhaitez utiliser.

La structure de la règle est la suivante.

{
    "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 prend actuellement en charge map-record-to-record et map-record-to-document en tant que seules valeurs valides pour le rule-action paramètre. Ces paramètres affectent les valeurs qui ne sont pas exclues de la liste d’attributs exclude-columns. Les map-record-to-document valeurs map-record-to-record et indiquent le mode de AWS DMS gestion de ces enregistrements par défaut. Ces valeurs n'affectent en aucune façon les mappages d'attributs.

Utilisez map-record-to-record lors d'une migration d'une base de données relationnelle vers une rubrique Kafka. Ce type de règle utilise la valeur taskResourceId.schemaName.tableName de la base de données relationnelle comme clé de partition dans la rubrique Kafka, et crée un attribut pour chaque colonne dans la base de données source.

Lorsque vous utilisez map-record-to-record, notez ce qui suit :

Une manière de comprendre map-record-to-record est de le voir en action. Dans cet exemple, imaginons que vous commencez avec une ligne de table d'une base de données relationnelle, présentant la structure et les données suivantes :

Pour migrer ces informations à partir d'un schéma nommé Test vers une rubrique Kafka, vous créez des règles pour mapper les données sur la rubrique cible. La règle suivante illustre ce mappage.

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

Compte tenu d’une rubrique Kafka et d’une clé de partition (dans ce cas, taskResourceId.schemaName.tableName), l’exemple ci-dessous illustre le format d’enregistrement résultant à l’aide de nos exemples de données dans la rubrique cible Kafka :

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

Restructuration de données avec le mappage d'attribut

Vous pouvez restructurer les données lors de leur migration vers une rubrique Kafka à l'aide d'un mappage d'attribut. Par exemple, vous pourriez vouloir regrouper plusieurs champs de la source en un seul champ dans la cible. Le mappage d'attribut suivant illustre comment restructurer les données.

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

Pour définir une valeur constante pour partition-key, spécifiez une valeur partition-key. Par exemple, vous pouvez le faire pour forcer le stockage de toutes les données dans une seule partition. Le mappage suivant illustre cette approche.

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

Réplication à plusieurs rubriques à l’aide du mappage d’objet

Par défaut, les AWS DMS tâches migrent toutes les données sources vers l'une des rubriques Kafka suivantes :

Avec les versions 3.4.6 et supérieures AWS DMS du moteur, vous pouvez utiliser l'kafka-target-topicattribut pour associer chaque table source migrée à une rubrique distincte. Par exemple, les règles de mappage d’objet suivantes migrent les tables sources Customer et Address vers les rubriques Kafka customer_topic et address_topic, respectivement. Dans le même temps, AWS DMS migre toutes les autres tables sources, y compris la Bills table du Test schéma, vers le sujet spécifié dans le point de terminaison cible.

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

En utilisant la réplication à plusieurs rubriques Kafka, vous pouvez regrouper et migrer les tables sources afin de séparer les rubriques Kafka à l’aide d’une seule tâche de réplication.

Format de message pour Apache Kafka

La sortie JSON est simplement une liste de paires clé-valeur.

L’exemple de message JSON suivant illustre un message de type de données avec toutes les métadonnées supplémentaires.

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

L’exemple de message JSON suivant illustre un message de type contrôle.

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