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

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

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

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

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

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

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

AWS Database Migration Service 使用 JSON 将记录发布到 Kafka 主题。转换期间， AWS DMS 将每个记录从源数据库序列化到 JSON 格式的属性/值对。

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

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

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

使用传输层安全性协议（TLS）连接到 Kafka

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

将 SSL 加密与 Amazon MSK 或自管理 Kafka 集群结合使用

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

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

使用 SSL 身份验证

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

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

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

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

要使用 SASL，您需要在设置 Amazon MSK 集群时先创建一个安全的用户名和密码。有关如何为 Amazon MSK 集群设置安全用户名和密码的说明，请参阅《Amazon Managed Streaming for Apache Kafka 开发人员指南》中的为 Amazon 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 行的原始值

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

通过在转换规则中使用 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 字段。当您将 CDC 更新写入 Amazon S3 目标时，通过 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 作为目标时的限制 AWS Database Migration Service

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

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

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

AWS DMS 目前支持将map-record-to-record和map-record-to-document作为该rule-action参数的唯一有效值。这些设置会影响未作为 exclude-columns 属性列表一部分排除的值。map-record-to-record和map-record-to-document值指定默认情况下如何 AWS 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}"
                    }
                ]
            }
        }
    ]
}

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

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

在 3.4.6 及更高版本的 AWS DMS 引擎中，您可以使用kafka-target-topic属性将每个迁移的源表映射到单独的主题。例如，下面的对象映射规则分别将源表 Customer 和 Address 迁移到 Kafka 主题 customer_topic 和 address_topic。同时，将所有其他源表（包括Test架构中的Bills表） AWS 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"
   }
}