Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

以 Amazon Kinesis Data Streams 为接收器 Amazon Database Migration Service

您可以使用将数据迁移Amazon DMS到Amazon Kinesis 数据流。Amazon Kinesis Data Streams 服务的一部分。您可以使用 Kinesis 数据流实时收集和处理大型数据流记录。

数据流由分片组成 Kinesis 数据流。分片是流中数据记录的唯一标识序列。有关Amazon Kinesis Data Streams 开发人员指南中的分区，请参阅 Am azon Kinesis D ata Streams 开发人员指南中的分区。

Amazon Database Migration Service使用 JSON 将记录发布到 Kinesis 数据流。在转换期间，Amazon DMS 将每个记录从源数据库序列化到 JSON 格式的属性/值对或 JSON_UNFORMATTED 消息格式。JSON_UNFORMATTED 消息格式是带有换行符的单行 JSON 字符串。它允许Amazon Kinesis Data Firehose 将 Kinesis 数据传送到Amazon S3 目的地，然后使用包括Amazon Athena 在内的各种查询引擎进行查询。

您将使用对象映射将数据从支持的数据源迁移到目标流。使用对象映射，您确定如何在流中建立数据记录结构。您还可以为每个表定义一个分区键，Kinesis Data Streams 使用该分区键将数据分组到其分区中。

在 Kinesis Data Streams 目标端点上Amazon DMS创建表时，它创建的表数量与源数据库端点中的表一样多。 Amazon DMS还设置了多个 Kinesis Data Streams 参数值。创建表的成本取决于要迁移的数据量和表数。

以下示例显示了正在使用的kinesis-settings选项以及使用发出create-endpoint命令的示例Amazon 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

使用 ParallelApply* 任务设置时， partition-key-type 默认值是表的 primary-key，而不是 schema-name.table-name。

使用之前的图像查看 Kinesis 数据流作为目标的 CDC 行的原始值

在将 CDC 更新写入 Kinesis 等数据流目标时，可以在更新更改之前查看源数据库行的原始值。为此，Amazon DMS 根据源数据库引擎提供的数据填充更新事件的之前映像。

不同的源数据库引擎为之前映像提供不同的信息量：

要启用之前映像以便将源数据库中的原始值添加到 Amazon DMS 输出，请使用 BeforeImageSettings 任务设置或 add-before-image-columns 参数。此参数应用列转换规则。

BeforeImageSettings 使用从源数据库系统收集的值向每个更新操作添加一个新的 JSON 属性，如下所示。

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

对于 BeforeImageSettings 选项，以下条件适用：

使用之前映像转换规则

作为任务设置的替代方法，您可以使用 add-before-image-columns 参数，该参数应用列转换规则。使用此参数，您可以在 CDC 期间对数据流目标（如 Kinesis）启用映像前功能。

通过在转换规则中使用 add-before-image-columns，可以对之前映像结果应用更精细的控制。转换规则允许您使用对象定位器，该定位器允许您控制为规则选择的表。此外，您可以将转换规则链接在一起，这样可以将不同的规则应用于不同的表。然后，您可以操控使用其他规则生成的列。

包含列的 add-before-image-columns 参数的 transformation 规则类型必须提供一个 before-image-def 部分。下面是一个示例。

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

column-prefix 的值附加到列名称前面，column-prefix 的默认值为 BI_。column-suffix 的值将附加到列名称之后，默认值为空。不要同时将 column-prefix 和 column-suffix 设置为空字符串。

为 column-filter 选择一个值。要仅添加属于表主键一部分的列，请选择 pk-only。选择 non-lob 以仅添加不属于 LOB 类型的列。或者，选择 all 以添加任何具有之前映像值的列。

之前映像转换规则的示例

以下示例中的转换规则在目标中添加一个名为 BI_emp_no 的新列。因此，像 UPDATE employees SET emp_no = 3 WHERE emp_no = 1; 这样的语句用 1 填充 BI_emp_no 字段。当您向 Amazon S3 目标写入 CDC 更新时，该BI_emp_no列可以分辨出哪个原始行已更新。

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

有关使用 add-before-image-columns 规则操作的信息，请参阅 转换规则和操作。

以 Kinesis 数据流作为目标的先决条件 Amazon Database Migration Service

在将 Kinesis 数据流设置为目标之前Amazon DMS，请确保创建 IAM 角色。此角色必须允许Amazon DMS代入和授予对正在迁移到的 Kinesis 数据流的访问权限。以下 IAM 策略中显示了所需的最小访问权限集合。

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

        

您用于迁移到 Kinesis 数据流的角色必须具有以下权限。

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

以 Kinesis Data Streams 为接收者时的限制 Amazon Database Migration Service

使用 Kinesis Data Streams 作为目标时，存在以下限制：

使用对象映射将数据迁移到 Kinesis 数据流

Amazon DMS使用表映射规则将数据从源映射到目标 Kinesis 数据流。要将数据映射到目标流，您必须使用称为 object mapping 的表映射规则类型。您可以使用对象映射来定义源中的数据记录如何映射到发布到 Kinesis 数据流的数据记录。

除了分区键外，Kinesis 数据流没有其他预设结构。在对象映射规则中，数据记录的 partition-key-type 的可能值为 schema-table、transaction-id、primary-key、constant 和 attribute-name。

要创建对象映射规则，您应将 rule-type 指定为 object-mapping。此规则指定您要使用的对象映射的类型。

规则的结构如下所示。

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

Amazon DMS 目前支持 map-record-to-record 和 map-record-to-document 作为 rule-action 参数的唯一有效值。map-record-to-record 和 map-record-to-document 值指定 Amazon DMS 默认情况下对未作为 exclude-columns 属性列表的一部分排除的记录执行的操作。这些值不会以任何方式影响属性映射。

从关系数据库迁移到 Kinesis 数据流map-record-to-record时使用。此规则类型使用关系数据库中的taskResourceId.schemaName.tableName值作为 Kinesis 数据流中的分区键，并为源数据库中的每列创建一个属性。在使用 map-record-to-record 时，对于源表中未在 exclude-columns 属性列表中列出的任何列，Amazon DMS 将在目标流中创建对应的属性。不论是否在属性映射中使用源列，都会创建对应的属性。

map-record-to-document用于使用属性名称 “_doc” 将源列放入相应目标流中的单个平面文档中。 Amazon DMS将数据放入名为 “_doc” 的源上的单一平面地图中。此放置应用于源表中的未在 exclude-columns 属性列表中列出的任何列。

了解 map-record-to-record 的一种方法是在操作时加以观察。对于本示例，假定您使用关系数据库表行开始处理，该行具有以下结构和数据。

要将此信息从名为 Kinesis 数据流的架构迁移Test到 Kinesis 数据流，需要创建规则将数据映射到目标流。以下规则对此映射进行了说明。

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

下图说明了 Kinesis 数据流中生成的记录格式：

但是，假设您使用相同的规则，但将rule-action参数更改为map-record-to-document并排除某些列。以下规则对此映射进行了说明。

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

在这种情况下，exclude-columns参数、FirstNameStoreId和DateOfBirth中未列出的列将映射到_doc。LastName以下说明了生成的记录格式。

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

使用属性映射调整数据结构

在使用属性映射将数据迁移到 Kinesis 数据流时，您可以重构数据。例如，您可能希望将源中的多个字段合并到目标中的单个字段中。以下属性映射说明如何调整数据结构。

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

要为 partition-key 设置常量值，请指定 partition-key 值。例如，您可以执行此操作来强制将所有数据存储在一个分片内。以下映射说明了此方法。

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

Kinesis Data Streams 的消息格式

JSON 输出只是键值对的列表。JSON_UNFORMATTED 消息格式是带有换行符的单行 JSON 字符串。

Amazon DMS提供以下保留字段，以便更轻松地使用 Kinesis Data Streams 中的数据：