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 for Apache MSK (Amazon) à utiliser comme AWS DMS cible. Amazon MSK est un service de streaming Apache Kafka entièrement géré qui simplifie la mise en œuvre et la gestion des instances Apache Kafka. Il fonctionne avec les versions open source d'Apache Kafka, et vous accédez aux MSK instances Amazon 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 MSK instance Amazon, un cluster exécuté sur une EC2 instance Amazon ou un cluster sur site. Une MSK instance Amazon ou un cluster sur une EC2 instance Amazon peut se trouver dans la même instance VPC ou dans une autre instance. 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 MSK cluster Amazon, assurez-vous que son groupe de sécurité autorise l'accès depuis votre instance de réplication. Pour plus d'informations sur la modification du groupe de sécurité d'un MSK cluster Amazon, consultez Modifier le groupe de sécurité d'un MSK cluster Amazon.

AWS Database Migration Service publie des enregistrements sur un sujet Kafka en utilisantJSON. Au cours de la conversion, AWS DMS sérialise chaque enregistrement de la base de données source en 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 de Transport Layer Security () TLS

Un cluster Kafka accepte les connexions sécurisées à l'aide de Transport Layer Security (TLS). AvecDMS, vous pouvez utiliser l'une des trois options de protocole de sécurité suivantes pour sécuriser une connexion à un terminal Kafka.

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

Vous pouvez utiliser SSL le chiffrement pour sécuriser la connexion d'un terminal à Amazon MSK ou à un cluster Kafka autogéré. Lorsque vous utilisez la méthode d'authentification par SSL chiffrement, 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'SSLauthentification

Vous pouvez utiliser l'SSLauthentification pour sécuriser la connexion d'un terminal à Amazon MSK ou à un cluster Kafka autogéré.

Pour activer l'authentification et le chiffrement du client à l'aide de l'SSLauthentification pour se connecter à AmazonMSK, procédez comme suit :

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

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

Pour l'utiliserSASL, vous devez d'abord créer un nom d'utilisateur et un mot de passe sécurisés lorsque vous configurez votre MSK cluster Amazon. Pour savoir comment configurer un nom d'utilisateur et un mot de passe sécurisés pour un MSK cluster Amazon, consultez la section Configuration SASL et SCRAM authentification pour un MSK cluster Amazon 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'ils correspondent au nom d'utilisateur et au mot de passe sécurisés que vous avez créés lors de la première configuration de votre MSK cluster Amazon, comme illustré dans l'JSONexemple suivant.

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

Utilisation d'une image précédente pour afficher les valeurs originales des CDC lignes pour Apache Kafka en tant que cible

Lorsque vous rédigez CDC des mises à jour sur une cible de diffusion de données telle que 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.

BeforeImageSettingsajoute un nouvel JSON attribut à chaque opération de mise à jour avec les valeurs collectées dans le 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 l'activer avant de créer des images 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 de n'ajouter que des colonnes qui ne sont pas de LOB type. 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 rédigez CDC des mises à jour pour les cibles Amazon S3, la BI_emp_no colonne 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 au 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 comment ces AWS DMS enregistrements sont gérés 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

Le JSON résultat est simplement une liste de paires clé-valeur.

L'exemple de JSON message 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 JSON message 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"
   }
}