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

使用 Amazon Kinesis Data Streams 作为目标 AWS Database Migration Service

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

Kinesis 数据流由分片组成。分片是流中数据记录的唯一标识序列。有关 Amazon Kinesis Data Streams 中的分片的更多信息，请参阅《Amazon Kinesis Data Streams 开发人员指南》中的分片。

AWS Database Migration Service 使用将记录发布到 Kinesis 数据流。JSON在转换过程中，将源数据库中的每条记录 AWS DMS 序列化为JSON格式或 JSON _ UNFORMATTED 消息格式的属性值对。JSON_ UNFORMATTED 消息格式是带有换行分隔符的单行JSON字符串。它允许 Amazon Data Firehose 将 Kinesis 数据传递到 Amazon S3 目标，然后使用各种查询引擎（包括 Amazon Athena）进行查询。

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

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

以下示例显示在使用 AWS CLI发出的示例 create-endpoint 命令中使用的 kinesis-settings 选项。

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行的原始值

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

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

要启用之前映像以便将源数据库中的原始值添加到 AWS 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 参数，该参数应用列转换规则。使用此参数，您可以在成像之前在数据流目标（如 Kinesis）CDC上启用。

通过在转换规则中使用 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 数据流作为目标的先决条件 AWS Database Migration Service

IAM用于使用 Kinesis 数据流作为目标的角色 AWS Database Migration Service

在将 Kinesis 数据流设置为目标之前 AWS DMS，请务必创建IAM角色。此角色必须允许 AWS 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 数据流作为目标 AWS Database Migration Service

在 3.4.7 及更高 AWS DMS 版本中，要连接到 Kinesis 终端节点，必须执行以下操作之一：

使用 Kinesis Data Streams 作为目标时的限制 AWS Database Migration Service

将 Kinesis Data Streams 作为目标时存在以下限制：

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

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

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

从关系数据库迁移到 Kinesis 数据流时使用 map-record-to-record。此规则类型使用关系数据库的 taskResourceId.schemaName.tableName 值作为 Kinesis 数据流中的分区键，并为源数据库中的每个列创建一个属性。

使用 map-record-to-record 时请注意：

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

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

要将此信息从名为 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 参数中未列出的列 FirstName、LastName、StoreId 和 DateOfBirth 将映射到 _doc。以下内容说明生成的记录格式。

       {
            "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字符串。

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