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

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

使用 Apache Kafka 作为目标 Amazon Database Migration Service

您可以使用将数据迁移 Amazon DMS 到 Apache Kafka 集群。Apache Kafka 是一个分布式流式处理平台。您可以使用 Apache Kafka 实时提取和处理流数据。

Amazon 还提供适用于 Apache Kafka 的 Apache Streaming（MSK亚马逊）的亚马逊托管流媒体作为目标 Amazon DMS 。亚马逊MSK是一项完全托管的 Apache Kafka 流媒体服务，可简化 Apache Kafka 实例的实施和管理。它适用于开源 Apache Kafka 版本，你可以像任何 Apache Kafka MSK 实例一样作为 Amazon DMS 目标访问亚马逊实例。有关更多信息，请参阅什么是亚马逊MSK？ 在《适用于 Apache 的亚马逊托管流媒体 Kafka 开发者指南》中。

Kafka 集群将记录流存储在称为主题的类别中，这些类别被划分为多个分区。分区是主题中唯一标识的数据记录（消息）序列。分区可分布在集群中的多个代理之间，以支持并行处理某个主题的记录。有关 Apache Kafka 中主题和分区及其分布的更多信息，请参阅主题和日志和分布。

您的 Kafka 集群可以是 Amazon MSK 实例、在 Amazon EC2 实例上运行的集群或本地集群。Amazon MSK 实例或 Amazon EC2 实例上的集群可以位于相同VPC或不同的集群中。如果您的集群位于本地，可以为复制实例使用自己的本地名称服务器以解析集群的主机名。有关在复制实例上设置名称服务器的信息，请参阅 使用您自己的本地名称服务器。有关设置网络的更多信息，请参阅为复制实例设置网络。

使用 Amazon MSK 集群时，请确保其安全组允许从您的复制实例进行访问。有关更改 Amazon MSK 集群安全组的信息，请参阅更改 Amazon MSK 集群的安全组。

Amazon Database Migration Service 使用JSON向 Kafka 主题发布记录。在转换过程中，将源数据库中的每条记录 Amazon DMS 序列化为格式的属性值对。JSON

要将数据从支持的任何数据源迁移到目标 Kafka 集群，您应使用对象映射。使用对象映射，可确定如何在目标主题中建立数据记录结构。您还可以为每个表定义分区键，Apache Kafka 使用该键将数据分组为分区。

目前， Amazon DMS 支持每个任务使用一个主题。对于包含多个表的单个任务，所有消息都将转到同一个主题。每条消息都包含一个标识目标架构和表的元数据部分。 Amazon DMS 3.4.6 及更高版本支持使用对象映射进行多主题复制。有关更多信息，请参阅 使用对象映射进行多主题复制。

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

使用传输层安全性连接到 Kafka () TLS

Kafka 集群使用传输层安全 (TLS) 接受安全连接。使用DMS，您可以使用以下三个安全协议选项中的任何一个来保护 Kafka 端点连接。

在 Amazon MSK 或自行管理的 Kafka 集群中使用SSL加密

您可以使用SSL加密来保护与 Amazon MSK 或自行管理的 Kafka 集群的终端节点连接。使用SSL加密身份验证方法时，客户端会通过服务器的证书来验证服务器的身份。然后在服务器和客户端之间建立加密连接。

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

使用 SSL 身份验证

您可以使用SSL身份验证来保护与 Amazon MSK 或自行管理的 Kafka 集群的终端节点连接。

要使用SSL身份验证启用客户端身份验证和加密以连接到 AmazonMSK，请执行以下操作：

使用 SASL-SSL 身份验证连接到 Amazon MSK

简单身份验证和安全层 (SASL) 方法使用用户名和密码来验证客户端身份，并在服务器和客户端之间建立加密连接。

要使用SASL，您需要在设置 Amazon MSK 集群时先创建一个安全的用户名和密码。有关如何为亚马逊MSK集群设置安全用户名和密码的说明，请参阅《适用于 Apache Kafka 的 Apache Managed Streaming 开发者指南》中的亚马逊MSK集群设置SASL/SCRAM身份验证。

然后，在创建 Kafka 目标端点时，使用 sasl-ssl 选项设置安全协议端点设置（SecurityProtocol）。您还可以设置 SaslUsername 和 SaslPassword 选项。请确保它们与您在首次设置 Amazon MSK 集群时创建的安全用户名和密码一致，如以下JSON示例所示。

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

使用之前的图像查看 Apache Kafka 作为目标的CDC行的原始值

在向 Kafka 等数据流目标写入CDC更新时，可以在更新更改之前查看源数据库行的原始值。为此，请根据源数据库引擎提供的数据 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 参数，该参数应用列转换规则。使用此参数，您可以在映像之前在数据流目标（如 Kafka）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 规则操作的信息，请参阅 转换规则和操作。

使用 Apache Kafka 作为目标时的限制 Amazon Database Migration Service

将 Apache Kafka 作为目标时存在以下限制：

使用对象映射将数据迁移到 Kafka 主题

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

除了具有分区键以外，Kafka 主题没有预设结构。

要创建对象映射规则，请将 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参数的唯一有效值。这些设置会影响未作为 exclude-columns 属性列表一部分排除的值。map-record-to-record和map-record-to-document值指定默认情况下如何 Amazon DMS 处理这些记录。这些值不会以任何方式影响属性映射。

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

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

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

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

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

给定 Kafka 主题和分区键（在本例中为 taskResourceId.schemaName.tableName），下面说明了使用 Kafka 目标主题中的示例数据生成的记录格式：

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

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

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

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

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

使用对象映射进行多主题复制

默认情况下， Amazon DMS 任务会将所有源数据迁移到以下 Kafka 主题之一：

在 3.4.6 及更高版本的 Amazon DMS 引擎中，您可以使用kafka-target-topic属性将每个迁移的源表映射到单独的主题。例如，下面的对象映射规则分别将源表 Customer 和 Address 迁移到 Kafka 主题 customer_topic 和 address_topic。同时，将所有其他源表（包括Test架构中的Bills表） Amazon DMS 迁移到目标终端节点中指定的主题。

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

通过使用 Kafka 多主题复制，您可以使用单个复制任务将源表分组并迁移到单独的 Kafka 主题。

Apache Kafka 的消息格式

JSON输出只是一个键值对列表。

以下JSON消息示例说明了包含所有其他元数据的数据类型消息。

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

以下JSON消息示例说明了控件类型消息。

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