本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

使用 Apache Kafka 做為 的目標 AWS Database Migration Service

您可以使用 AWS DMS 將資料遷移至 Apache Kafka 叢集。Apache Kafka 是一個分散式串流平台。您可以使用 Apache Kafka 來即時擷取和處理串流資料。

AWS 也提供 Amazon Managed Streaming for Apache Kafka (Amazon MSK) 做為 AWS DMS 目標使用。Amazon MSK 是一種全受管 Apache Kafka 串流服務，可簡化 Apache Kafka 執行個體的實作和管理。它適用於開放原始碼 Apache Kafka 版本，而且您可以像任何 Apache Kafka MSK執行個體一樣，將 Amazon 執行個體做為 AWS DMS 目標存取。如需詳細資訊，請參閱《Amazon Managed Streaming for Apache Kafka 開發人員指南》中的什麼是 AmazonMSK？。

Kafka 叢集會記錄串流儲存在稱為主題的類別裡，其中具有分割區。分割區是主題中資料記錄 (訊息) 的唯一識別序列。分割區可以分佈在叢集中的多個中介裝置，以啟用主題記錄的平行處理。如需有關主題和分割區及其在 Apache Kafka 中分佈的詳細資訊，請參閱主題和日誌以及分佈。

您的 Kafka 叢集可以是 Amazon MSK執行個體、在 Amazon EC2執行個體上執行的叢集，或內部部署叢集。Amazon MSK執行個體或 Amazon EC2執行個體上的叢集可以是相同VPC或不同的執行個體。如果您的叢集為內部部署，您可以針對複寫執行個體使用自己的內部部署名稱伺服器來解析叢集的主機名稱。如需為複寫執行個體設定名稱伺服器的相關資訊，請參閱 使用自己的內部部署名稱伺服器。如需設定網路的詳細資訊，請參閱 設定複寫執行個體的網路。

使用 Amazon MSK叢集時，請確定其安全群組允許從您的複寫執行個體存取 。如需變更 Amazon MSK叢集安全群組的詳細資訊，請參閱變更 Amazon MSK叢集的安全群組。

AWS Database Migration Service 使用 將記錄發佈至 Kafka 主題JSON。在轉換期間， AWS DMS 會將來源資料庫中的每個記錄序列化為JSON格式的屬性值對。

若要將資料從任何受支援的資料來源遷移到目標 Kafka 叢集，可以使用物件對應。您可以使用物件映射來決定如何建構目標主題中的資料記錄。您也可以定義每份資料表的分割區索引鍵，Apache Kafka 用它將資料分組為分割區。

目前， AWS DMS 支援每個任務的單一主題。對於具有多個資料表的單一任務，所有訊息都會傳送至同一主題。每個訊息都包含中繼資料區段，可識別目標結構描述和 table. AWS DMS versions 3.4.6 和更新版本，支援使用物件映射的多主題複寫。如需詳細資訊，請參閱使用物件映射的多主題複製。

在使用 ParallelApply* 任務設定時，partition-key-type 預設是資料表的 primary-key，而非 schema-name.table-name。

使用 Transport Layer Security 連線至 Kafka (TLS)

Kafka 叢集接受使用 Transport Layer Security () 的安全連線TLS。使用 DMS時，您可以使用下列三個安全通訊協定選項中的任何一個來保護 Kafka 端點連線。

搭配 Amazon MSK或自我管理的 Kafka 叢集使用SSL加密

您可以使用SSL加密來保護 Amazon MSK或自我管理 Kafka 叢集的端點連線。當您使用SSL加密身分驗證方法時，用戶端會透過伺服器的憑證驗證伺服器的身分。接著在伺服器和用戶端之間建立加密連線。

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

使用SSL身分驗證

您可以使用SSL身分驗證來保護 Amazon MSK或自我管理 Kafka 叢集的端點連線。

若要使用身分驗證啟用用戶端SSL身分驗證和加密以連線至 Amazon MSK，請執行下列動作：

使用 SASL-SSL 身分驗證連線到 Amazon MSK

Simple Authentication and Security Layer (SASL) 方法使用使用者名稱和密碼來驗證用戶端身分，並在伺服器和用戶端之間進行加密連線。

若要使用 SASL，請先在設定 Amazon MSK叢集時建立安全的使用者名稱和密碼。如需如何為 Amazon MSK叢集設定安全使用者名稱和密碼的說明，請參閱《Amazon Managed Streaming for Apache Kafka 開發人員指南》中的設定SASL/SCRAM驗證 Amazon MSK叢集。

接著，當您建立 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 參數做為任務設定的替代方式，它會套用欄轉換規則。使用此參數，您可以在 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 欄位。當您將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 主題：

使用 AWS DMS 引擎 3.4.6 版和更新版本時，您可以使用 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"
   }
}