Utilisation d'Amazon Kinesis Data Streams comme cible pour AWS Database Migration Service

Vous pouvez l'utiliser AWS DMS pour migrer des données vers un flux de données Amazon Kinesis. Les flux de données Amazon Kinesis font partie du service Amazon Kinesis Data Streams. Vous pouvez utiliser des flux de données Kinesis pour collecter et traiter des flux volumineux d’enregistrements de données en temps réel.

Un flux de données Kinesis se compose de partitions. Les partitions sont des séquences d'enregistrements de données identifiées de manière unique dans un flux. Pour plus d’informations sur les partitions dans Amazon Kinesis Data Streams, consultez Partition dans le Guide du développeur Amazon Kinesis Data Streams.

AWS Database Migration Service publie des enregistrements dans un flux de données Kinesis à 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 ou un format de message JSON_UNFORMATED. Un format de message JSON_UNFORMATED est une chaîne JSON à une seule ligne avec un délimiteur de nouvelle ligne. Il permet à Amazon Data Firehose de fournir des données Kinesis à une destination Amazon S3, puis de les interroger à l'aide de différents moteurs de requête, dont Amazon Athena.

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

Lors de la AWS DMS création de tables sur un point de terminaison cible Kinesis Data Streams, il crée autant de tables que dans le point de terminaison de la base de données source. AWS DMS définit également plusieurs valeurs de paramètres Kinesis Data Streams. Le coût de création de la table dépend de la quantité de données et du nombre de tables à migrer.

Note

L'option Mode SSL de la AWS DMS console ou de l'API ne s'applique pas à certains services de streaming de données et NoSQL tels que Kinesis et DynamoDB. Ils sont sécurisés par défaut, ce qui AWS DMS montre que le paramètre du mode SSL est égal à aucun (mode SSL = aucun). Vous n’avez pas besoin de fournir de configuration supplémentaire pour que votre point de terminaison utilise le protocole SSL. Par exemple, lorsque Kinesis est utilisé comme point de terminaison cible, il est sécurisé par défaut. Tous les appels d'API adressés à Kinesis utilisent le protocole SSL, il n'est donc pas nécessaire d'ajouter une option SSL supplémentaire sur le AWS DMS terminal. Vous pouvez placer et récupérer des données en toute sécurité via des points de terminaison SSL à l’aide du protocole HTTPS, utilisé par AWS DMS par défaut lors de la connexion à un flux de données Kinesis.

Paramètres de point de terminaison Kinesis Data Streams

Lorsque vous utilisez les points de terminaison cibles de Kinesis Data Streams, vous pouvez obtenir les détails des transactions et des contrôles à l'aide de l'option KinesisSettings de l'API. AWS DMS

Vous pouvez définir les paramètres de connexion de l’une des manières suivantes :

• Dans la AWS DMS console, à l'aide des paramètres du point de terminaison.

• Dans la CLI, en utilisant l'kinesis-settingsoption de la CreateEndpointcommande.

Dans l’interface CLI, utilisez les paramètres de demande de l’option kinesis-settings suivants :

Note

La prise en charge des paramètres de point de terminaison IncludeNullAndEmpty est disponible dans AWS DMS versions 3.4.1 et ultérieures. Mais la prise en charge des autres paramètres de point de terminaison suivants pour les cibles Kinesis Data Streams est disponible dans AWS DMS.

• MessageFormat : format de sortie pour les enregistrements créés sur le point de terminaison. Le format du message est JSON (par défaut) ou JSON_UNFORMATTED (une seule ligne sans onglet).

• IncludeControlDetails : affiche des informations de contrôle détaillées pour la définition de table, la définition de colonne et les modifications de table et de colonne dans la sortie du message Kinesis. L’argument par défaut est false.

• IncludeNullAndEmpty : inclut les colonnes NULL et vides dans la cible. L’argument par défaut est false.

• IncludePartitionValue : affiche la valeur de partition dans la sortie du message Kinesis, sauf si le type de partition est schema-table-type. L’argument par défaut est false.

• IncludeTableAlterOperations : inclut toutes les opérations DDL (Data Definition Language) qui modifient la table dans les données de contrôle, telles que rename-table, drop-table, add-column, drop-column et rename-column. L’argument par défaut est false.

• IncludeTransactionDetails : fournit des informations détaillées sur les transactions à partir de la base de données source. Ces informations comprennent un horodatage de validation, une position de journal et des valeurs pour transaction_id, previous_transaction_id et transaction_record_id (le décalage d'enregistrement dans une transaction). L’argument par défaut est false.

• PartitionIncludeSchemaTable : préfixe les noms de schéma et de table aux valeurs de partition, lorsque le type de partition est primary-key-type. Cela augmente la distribution des données entre les fragments Kinesis. Par exemple, supposons qu'un schéma SysBench comporte des milliers de tables et que chaque table n'ait qu'une plage limitée pour une clé primaire. Dans ce cas, la même clé primaire est envoyée à partir de milliers de tables vers le même fragment, ce qui provoque une limitation. L’argument par défaut est false.

L’exemple suivant illustre l’option kinesis-settings utilisée avec un exemple de commande create-endpoint émise à l’aide d’ 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

Paramètres de tâche de chargement complet multithread

Pour accélérer le transfert, AWS DMS prend en charge le chargement complet multithread vers une instance cible Kinesis Data Streams. DMS prend en charge ce traitement multithread avec des paramètres de tâche, notamment les suivants :

• MaxFullLoadSubTasks : utilisez cette option pour indiquer le nombre maximal de tables sources à charger en parallèle. DMS charge chaque table dans sa table cible Kinesis correspondante à l’aide d’une sous-tâche dédiée. La valeur par défaut est 8 ; la valeur maximale 49.

• ParallelLoadThreads— Utilisez cette option pour spécifier le nombre de threads AWS DMS utilisés pour charger chaque table dans sa table cible Kinesis. La valeur maximale pour une cible Kinesis Data Streams est de 32. Vous pouvez demander une augmentation de cette limite maximale.

• ParallelLoadBufferSize : utilisez cette option pour spécifier le nombre maximal d’enregistrements à stocker dans la mémoire tampon utilisée par les threads de chargement parallèles pour charger les données dans la cible Kinesis. La valeur par défaut est 50. La valeur maximale est 1 000. Utilisez ce paramètre avec ParallelLoadThreads. ParallelLoadBufferSize est valide uniquement dans le cas de plusieurs threads.

• ParallelLoadQueuesPerThread : utilisez cette option pour spécifier le nombre de files d’attente auxquelles chaque thread simultané accède pour extraire les enregistrements de données des files d’attente et générer un chargement par lots pour la cible. La valeur par défaut est 1. Toutefois, pour les cibles Kinesis de différentes tailles de charge utile, la plage valide est comprise entre 5 et 512 files d’attente par thread.

Paramètres de tâche de chargement CDC multithread

Vous pouvez utiliser les paramètres de tâche afin d’améliorer les performances de la capture des données de modification (CDC) pour les points de terminaison cibles de streaming des données en temps réel, comme Kinesis et modifier le comportement de l’appel d’API PutRecords. Pour ce faire, vous pouvez spécifier le nombre de threads simultanés, les files d'attente par thread et le nombre d'enregistrements à stocker dans un tampon à l'aide de la tâche ParallelApply*. Par exemple, supposons que vous souhaitiez effectuer un chargement CDC et appliquer 128 threads en parallèle. Vous souhaitez également accéder à 64 files d'attente par thread, avec 50 enregistrements stockés par tampon.

Pour améliorer les performances du CDC, AWS DMS prend en charge les paramètres de tâche suivants :

• ParallelApplyThreads— Spécifie le nombre de threads simultanés AWS DMS utilisés lors d'un chargement CDC pour transférer des enregistrements de données vers un point de terminaison cible Kinesis. La valeur par défaut est zéro (0) et la valeur maximale est 32.

• ParallelApplyBufferSize : spécifie le nombre maximal d’enregistrements à stocker dans chaque file d’attente de mémoire tampon pour que les threads simultanés soient transférés vers un point de terminaison cible Kinesis lors d’un chargement CDC. La valeur par défaut est 100 et la valeur maximale est 1 000. Utilisez cette option lorsque ParallelApplyThreads spécifie plusieurs threads.

• ParallelApplyQueuesPerThread : spécifie le nombre de files d’attente auxquelles chaque thread accède pour extraire les enregistrements de données des files d’attente et générer un chargement par lots pour un point de terminaison Kinesis pendant la CDC. La valeur par défaut est 1 et la valeur maximale est 512.

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.

Utilisation d’une image antérieure pour afficher les valeurs d’origine des lignes CDC pour un flux de données Kinesis en tant que cible

Lorsque vous écrivez des mises à jour de CDC sur une cible de diffusion de données comme Kinesis, 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 :

• Oracle met uniquement à jour des colonnes si elles changent.

• PostgreSQL fournit uniquement des données pour les colonnes qui font partie de la clé primaire (modifiée ou non). Pour fournir des données pour toutes les colonnes (modifiées ou non), vous devez définir REPLICA_IDENTITY sur FULL au lieu de DEFAULT. Notez que vous devez choisir la valeur de REPLICA_IDENTITY avec soin pour chaque table. Si vous définissez REPLICA_IDENTITY sur FULL, toutes les valeurs de colonne sont écrites en continu dans la journalisation WAL. Cela peut entraîner des problèmes de performances ou de ressources avec les tables qui sont fréquemment mises à jour.

• MySQL fournit généralement des données pour toutes les colonnes, à l’exception des types de données BLOB et CLOB (modifiés ou non).

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

Note

S'applique uniquement BeforeImageSettings aux AWS DMS tâches contenant un composant CDC, telles que les tâches à chargement complet plus CDC (qui migrent les données existantes et répliquent les modifications en cours), ou aux tâches CDC uniquement (qui répliquent uniquement les modifications de données). N’appliquez pas les BeforeImageSettings aux tâches à pleine charge uniquement.

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

• Définissez l'option EnableBeforeImage sur true pour activer la génération d’image antérieure. L’argument par défaut est false.

• Utilisez l'option FieldName pour attribuer un nom au nouvel attribut JSON. Quand EnableBeforeImage est true, FieldName est obligatoire et ne peut pas être vide.

• L'option ColumnFilter spécifie une colonne à ajouter en utilisant la génération d’image antérieure. Pour ajouter uniquement des colonnes faisant partie des clés primaires de la table, utilisez la valeur par défaut, pk-only. Pour ajouter une colonne ayant une valeur d'image antérieure, utilisez all. Notez que l’image antérieure ne contient pas de colonnes avec des types de données LOB, tels que CLOB ou BLOB.

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

Note

Les cibles Amazon S3 ne prennent pas en charge BeforeImageSettings. Pour les cibles S3, utilisez uniquement la règle de transformation add-before-image-columns à effectuer avant la génération d’images pendant la CDC.

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 streaming de données comme Kinesis.

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.

Note

N'utilisez pas le paramètre add-before-image-columns avec le paramètre de tâche BeforeImageSettings dans la même tâche. N’utilisez pas les deux, pour une seule tâche.

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.

Conditions préalables à l'utilisation d'un flux de données Kinesis comme cible pour AWS Database Migration Service

Rôle IAM pour l'utilisation d'un flux de données Kinesis comme cible pour AWS Database Migration Service

Avant de configurer un flux de données Kinesis comme cible AWS DMS, assurez-vous de créer un rôle IAM. Ce rôle doit permettre d' AWS DMS assumer et d'accorder l'accès aux flux de données Kinesis vers lesquels la migration est en cours de migration. L'ensemble d'autorisations d'accès minimum est indiqué dans la stratégie IAM suivante.

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

        

Le rôle que vous utilisez pour la migration vers un flux de données Kinesis doit bénéficier des autorisations suivantes.

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

Accès à un flux de données Kinesis en tant que cible pour AWS Database Migration Service

Dans les AWS DMS versions 3.4.7 et supérieures, pour vous connecter à un point de terminaison Kinesis, vous devez effectuer l'une des opérations suivantes :

• Configurez DMS pour qu’il utilise les points de terminaison d’un VPC. Pour plus d’informations sur la configuration de DMS pour qu’il utilise les points de terminaison d’un VPC, consultez Configuration de points de terminaison de VPC en tant que points de terminaison sources et cibles AWS DMS.

• Configurez DMS pour qu’il utilise des routes publiques, c’est-à-dire qu’il rende votre instance de réplication publique. Pour plus d’informations sur les instances de réplication publiques, consultez Instances de réplication publiques et privées.

Limitations liées à l'utilisation de Kinesis Data Streams comme cible pour AWS Database Migration Service

Les limitations suivantes s’appliquent lorsque vous utilisez Kinesis Data Streams en tant que cible :

• AWS DMS publie chaque mise à jour d'un seul enregistrement de la base de données source sous la forme d'un enregistrement de données dans un flux de données Kinesis donné, quelles que soient les transactions. Toutefois, vous pouvez inclure des détails de transaction pour chaque enregistrement de données à l'aide des paramètres pertinents de l'API KinesisSettings.

• Le Mode LOB complet n'est pas pris en charge.

• La taille LOB maximale prise en charge est de 1 Mo.

• Kinesis Data Streams ne prend pas en charge la déduplication. Les applications qui consomment des données d'un flux doivent gérer les enregistrements en double. Pour plus d’informations, consultez Gestion des enregistrements en double dans le Guide du développeur Amazon Kinesis Data Streams.

• AWS DMS prend en charge les deux formes suivantes pour les clés de partition :

  • SchemaName.TableName : une combinaison du nom du schéma et du nom de la table.

  • ${AttributeName} : la valeur d'un des champs du fichier JSON, ou la clé primaire de la table dans la base de données source.

• Pour en savoir plus sur le chiffrement de vos données au repos dans Kinesis Data Streams, consultez Protection des données dans Kinesis Data Streams dans le Guide du développeur AWS Key Management Service .

• BatchApply n’est pas pris en charge pour un point de terminaison Kinesis. L’utilisation de l’application par lots (par exemple, le paramètre de tâche de métadonnées cible BatchApplyEnabled) pour une cible Kinesis peut entraîner une perte de données.

• Les cibles Kinesis ne sont prises en charge que pour un flux de données Kinesis dans le même AWS compte et le même Région AWS que l'instance de réplication.

• Lors de la migration depuis une source MySQL, les BeforeImage données n'incluent pas les types de données CLOB et BLOB. Pour de plus amples informations, veuillez consulter Utilisation d’une image antérieure pour afficher les valeurs d’origine des lignes CDC pour un flux de données Kinesis en tant que cible.

• AWS DMS ne prend pas en charge la migration de valeurs de type de BigInt données comportant plus de 16 chiffres. Pour contourner cette limitation, vous pouvez utiliser la règle de transformation suivante pour convertir la colonne BigInt en chaîne. Pour plus d’informations sur les règles de transformation, consultez Règles et actions de transformation.

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

Utilisation du mappage d’objet pour migrer les données vers un flux de données Kinesis

AWS DMS utilise des règles de mappage de tables pour mapper les données de la source vers le flux de données Kinesis cible. Pour mapper des données vers un flux cible, vous utilisez un type de règles de mappage de tables qu'on appelle le mappage d'objet. Utilisez le mappage d’objet pour définir la façon dont les enregistrements de données de la source sont mappés aux enregistrements de données publiés dans le flux de données Kinesis.

Les flux de données Kinesis ne disposent pas d’une structure prédéfinie autre que le fait d’avoir une clé de partition. Dans une règle de mappage d'objets, les valeurs possibles de partition-key-type pour un enregistrement de données sont schema-table, transaction-id, primary-key, constant et attribute-name.

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 un flux de données Kinesis. Ce type de règle utilise la valeur taskResourceId.schemaName.tableName de la base de données relationnelle comme clé de partition dans le flux de données Kinesis et crée un attribut pour chaque colonne de la base de données source.

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

• Ce paramètre n’affecte que les colonnes exclues par la liste exclude-columns.

• Pour chacune de ces colonnes, AWS DMS crée un attribut correspondant dans le sujet cible.

• AWS DMS crée cet attribut correspondant, que la colonne source soit utilisée ou non dans un mappage d'attributs.

Utilisez map-record-to-document pour placer des colonnes sources dans un document unique à plat du flux cible approprié en utilisant le nom d’attribut « _doc ». AWS DMS place les données dans un mappage unique à plat sur la source appelée « _doc ». Ce placement s'applique à toute colonne de la table source non listée dans la liste d'attributs exclude-columns.

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 :

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

Pour migrer ces informations d’un schéma nommé Test vers un flux de données Kinesis, vous créez des règles pour mapper les données au flux 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": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

L’exemple qui suit illustre le format d’enregistrement résultant dans le flux de données Kinesis :

• StreamName: XXX

• PartitionKey: Test.Customers //SchmaName.TableName

• Data : //The following JSON message

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

Supposons toutefois que vous utilisiez les mêmes règles, mais que vous redéfinissiez le paramètre rule-action sur map-record-to-document et que vous excluiez certaines colonnes. 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": "DefaultMapToKinesis",
			"rule-action": "map-record-to-document",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "Customers"
			},
			"mapping-parameters": {
				"exclude-columns": [
					"homeaddress",
					"homephone",
					"workaddress",
					"workphone"
				]
			}
		}
	]
}                

Dans ce cas, les colonnes non répertoriées dans le paramètre exclude-columns, FirstName, LastName, StoreId et DateOfBirth, sont mappées à _doc. L’exemple qui suit illustre le format d’enregistrement résultant.

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

Restructuration de données avec le mappage d'attribut

Vous pouvez restructurer les données lors de leur migration vers un flux de données Kinesis à 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": "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}"
                            }
                        }
                    }
                ]
            }
        }
    ]
}

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

Note

La valeur partition-key d'un enregistrement de contrôle correspondant à une table spécifique est TaskId.SchemaName.TableName. La valeur partition-key d'un enregistrement de contrôle correspondant à une tâche spécifique est le TaskId de cet enregistrement. La spécification d'une valeur partition-key dans le mappage d'objet n'a aucun impact sur la partition-key d'un enregistrement de contrôle.

Format de message pour Kinesis Data Streams

La sortie JSON est simplement une liste de paires clé-valeur. Un format de message JSON_UNFORMATED est une chaîne JSON à une seule ligne avec un délimiteur de nouvelle ligne.

AWS DMS fournit les champs réservés suivants pour faciliter la consommation des données issues des Kinesis Data Streams :

RecordType

Les enregistrements peuvent être de type Données ou Contrôle. Les enregistrements de données représentent les lignes réelles de la source. Les enregistrements de contrôle sont destinés à des événements importants dans le flux, par exemple un redémarrage de la tâche.

Opération

Pour les enregistrements de données, l'opération peut être load,insert, update ou delete.

Pour les enregistrements de contrôle, l’opération peut être create-table, rename-table, drop-table, change-columns, add-column, drop-column, rename-column ou column-type-change.

SchemaName

Schéma source de l'enregistrement. Ce champ peut être vide pour un enregistrement de contrôle.

TableName

Table source de l'enregistrement. Ce champ peut être vide pour un enregistrement de contrôle.

Horodatage

Horodatage de la construction du message JSON. Le champ est formaté selon le format ISO 8601.