Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Verwendung von Apache Kafka als Ziel für AWS Database Migration Service

Sie können es verwenden AWS DMS , um Daten zu einem Apache Kafka-Cluster zu migrieren. Apache Kafka ist eine verteilte Streaming-Plattform. Mit Apache Kafka können Sie Streaming-Daten in Echtzeit erfassen und verarbeiten.

AWS bietet auch Amazon Managed Streaming for Apache Kafka (AmazonMSK) zur Verwendung als AWS DMS Ziel an. Amazon MSK ist ein vollständig verwalteter Apache Kafka-Streaming-Service, der die Implementierung und Verwaltung von Apache Kafka-Instances vereinfacht. Es funktioniert mit Open-Source-Versionen von Apache Kafka, und Sie greifen genau wie jede Apache Kafka-Instance auf MSK Amazon-Instances als AWS DMS Ziele zu. Weitere Informationen finden Sie unter Was ist AmazonMSK? im Amazon Managed Streaming for Apache Kafka Developer Guide.

Ein Kafka-Cluster speichert Streams von Datensätzen in Kategorien, die als Themen bezeichnet und in Partitionen unterteilt werden. Partitionen sind eindeutig identifizierte Sequenzen von Datensätzen (Nachrichten) in einem Thema. Partitionen können über mehrere Broker in einem Cluster verteilt werden, um die parallele Verarbeitung der Datensätze eines Themas zu ermöglichen. Weitere Informationen zu Themen und Partitionen und ihrer Verteilung in Apache Kafka finden Sie unter Themen und Protokolle und Verteilung.

Ihr Kafka-Cluster kann entweder eine MSK Amazon-Instance, ein Cluster, der auf einer EC2 Amazon-Instance läuft, oder ein lokaler Cluster sein. Eine MSK Amazon-Instance oder ein Cluster auf einer EC2 Amazon-Instance kann sich in derselben VPC oder einer anderen Instance befinden. Wenn es sich um einen On-Premises-Cluster handelt, können Sie Ihren eigenen On-Premises-Namensserver für Ihre Replikations-Instance verwenden, um den Host-Namen des Clusters aufzulösen. Informationen zum Einrichten eines Namensservers für Ihre Replikations-Instance finden Sie unter Verwenden Ihres eigenen Vor-Ort-Nameservers. Weitere Informationen zum Einrichten eines Netzwerks finden Sie unter Einrichten eines Netzwerks für eine Replikations-Instance.

Wenn Sie einen MSK Amazon-Cluster verwenden, stellen Sie sicher, dass dessen Sicherheitsgruppe den Zugriff von Ihrer Replikationsinstanz aus ermöglicht. Informationen zum Ändern der Sicherheitsgruppe für einen MSK Amazon-Cluster finden Sie unter Ändern der Sicherheitsgruppe eines MSK Amazon-Clusters.

AWS Database Migration Service veröffentlicht Datensätze zu einem Kafka-Thema mitJSON. AWS DMS Serialisiert während der Konvertierung jeden Datensatz aus der Quelldatenbank in ein Attribut-Wert-Paar im Format. JSON

Zum Migrieren Ihrer Daten von einer unterstützten Datenquelle zu einem Kafka-Ziel-Cluster verwenden Sie die Objektzuweisung. Mit der Objektzuweisung bestimmen Sie, wie die Datensätze im Zielthema strukturiert werden sollen. Außerdem definieren Sie einen Partitionsschlüssel für jede Tabelle, die Apache Kafka zum Gruppieren der Daten in seine Partitionen verwendet.

AWS DMS Unterstützt derzeit ein einzelnes Thema pro Aufgabe. Bei einer einzelnen Aufgabe mit mehreren Tabellen beziehen sich alle Nachrichten auf ein einzelnes Thema. Jede Nachricht enthält einen Metadaten-Abschnitt, der das Zielschema und die Zieltabelle identifiziert. AWS DMS Versionen 3.4.6 und höher unterstützen die themenübergreifende Replikation mithilfe von Objektzuordnung. Weitere Informationen finden Sie unter Replikation für mehrere Themen mithilfe der Objektzuweisung.

Wenn Sie ParallelApply*-Aufgabeneinstellungen verwenden, ist der primary-key der Tabelle der partition-key-type-Standardwert, nicht schema-name.table-name.

Mit Transport Layer Security () eine Verbindung zu Kafka herstellen TLS

Ein Kafka-Cluster akzeptiert sichere Verbindungen mithilfe von Transport Layer Security ()TLS. Mit DMS können Sie eine der folgenden drei Sicherheitsprotokolloptionen verwenden, um eine Kafka-Endpunktverbindung zu sichern.

SSLVerschlüsselung mit Amazon MSK oder einem selbstverwalteten Kafka-Cluster verwenden

Sie können SSL Verschlüsselung verwenden, um eine Endpunktverbindung zu Amazon MSK oder einem selbstverwalteten Kafka-Cluster zu sichern. Wenn Sie die SSL Verschlüsselungsauthentifizierungsmethode verwenden, validieren Clients die Identität eines Servers anhand des Serverzertifikats. Anschließend wird eine verschlüsselte Verbindung zwischen Server und Client hergestellt.

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

SSLAuthentifizierung verwenden

Sie können die SSL Authentifizierung verwenden, um eine Endpunktverbindung zu Amazon MSK oder einem selbstverwalteten Kafka-Cluster zu sichern.

Gehen Sie wie folgt vor, um die SSL Client-Authentifizierung und Verschlüsselung mithilfe der Authentifizierung für die Verbindung mit Amazon MSK zu aktivieren:

Verwenden von SASL — SSL Authentifizierung für die Verbindung zu Amazon MSK

Die Methode Simple Authentication and Security Layer (SASL) verwendet einen Benutzernamen und ein Passwort, um die Identität eines Kunden zu überprüfen, und stellt eine verschlüsselte Verbindung zwischen Server und Client her.

Für die Verwendung SASL erstellen Sie zunächst einen sicheren Benutzernamen und ein sicheres Passwort, wenn Sie Ihren MSK Amazon-Cluster einrichten. Eine Beschreibung, wie Sie einen sicheren Benutzernamen und ein sicheres Passwort für einen MSK Amazon-Cluster einrichten, finden Sie unter Setting SASL SCRAM up/authentication for a Amazon MSK Cluster im Amazon Managed Streaming for Apache Kafka Developer Guide.

Wenn Sie dann Ihren Kafka-Zielendpunkt erstellen, legen Sie die Sicherheitsprotokoll-Endpunkteinstellung (SecurityProtocol) mithilfe der Option sasl-ssl fest. Legen Sie auch die Optionen SaslUsername und SaslPassword fest. Stellen Sie sicher, dass diese mit dem sicheren Benutzernamen und dem Passwort übereinstimmen, die Sie bei der ersten Einrichtung Ihres MSK Amazon-Clusters erstellt haben, wie im folgenden JSON Beispiel gezeigt.

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

Verwenden Sie ein Vorher-Bild, um die ursprünglichen Werte von CDC Zeilen für Apache Kafka als Ziel anzuzeigen

Wenn Sie CDC Aktualisierungen in ein Datenstreaming-Ziel wie Kafka schreiben, können Sie die ursprünglichen Werte einer Quelldatenbankzeile anzeigen, bevor sie durch ein Update geändert werden. Um dies zu ermöglichen, AWS DMS füllt es ein Vorher-Bild von Aktualisierungsereignissen auf der Grundlage von Daten aus, die von der Quelldatenbank-Engine bereitgestellt werden.

Verschiedene Quelldatenbank-Engines liefern unterschiedliche Mengen an Informationen für ein Vorher-Abbild:

Verwenden Sie entweder die BeforeImageSettings-Aufgabeneinstellung oder den add-before-image-columns-Parameter, um die Erstellung von Vorher-Abbildern zum Hinzufügen von Originalwerten aus der Quelldatenbank zur AWS DMS -Ausgabe zu aktivieren. Dieser Parameter wendet eine Spalten-Transformationsregel an.

BeforeImageSettingsfügt jedem Aktualisierungsvorgang mit Werten, die aus dem Quelldatenbanksystem gesammelt wurden, ein neues JSON Attribut hinzu, wie im Folgenden gezeigt.

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

Für BeforeImageSettings-Optionen gilt Folgendes:

Verwenden einer Vorher-Abbild-Transformationsregel

Alternativ zu den Aufgabeneinstellungen können Sie den add-before-image-columns-Parameter verwenden, der eine Spalten-Transformationsregel anwendet. Mit diesem Parameter können Sie das Imaging CDC auf Datenstreaming-Zielen wie Kafka vor dem Start aktivieren.

Wenn Sie add-before-image-columns in einer Transformationsregel verwenden, können Sie eine feinere Steuerung der Ergebnisse für das Vorher-Abbild anwenden. Mit Transformationsregeln können Sie einen Objekt-Locator verwenden, der Ihnen die Kontrolle über die für die Regel ausgewählten Tabellen gibt. Außerdem können Sie Transformationsregeln miteinander verketten, wodurch verschiedene Regeln auf verschiedene Tabellen angewendet werden können. Anschließend können Sie die erzeugten Spalten mithilfe anderer Regeln bearbeiten.

Ein transformation-Regeltyp mit dem add-before-image-columns-Parameter für eine Spalte muss einen before-image-def-Abschnitt bereitstellen. Es folgt ein Beispiel.

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

Der Wert von column-prefix wird einem Spaltennamen vorangestellt, und der Standardwert von column-prefix ist BI_. Der Wert von column-suffix wird an den Spaltennamen angehängt, und der Standardwert ist leer. Setzen Sie nicht column-prefix und column-suffix auf leere Zeichenfolgen.

Wählen Sie einen Wert für column-filter. Wenn Sie nur Spalten hinzufügen möchten, die Teil der Primärschlüssel der Tabelle sind, wählen Sie pk-only . Wählen Sie ausnon-lob, dass nur Spalten hinzugefügt werden sollen, die nicht vom LOB Typ sind. Oder lassenall Sie eine Spalte hinzufügen, die einen Vorher-Abbild-Wert hat.

Beispiel für eine Vorher-Abbild-Transformationsregel

Die Transformationsregel im folgenden Beispiel fügt eine neue Spalte mit dem Namen BI_emp_no auf dem Ziel hinzu. Eine Anweisung wie UPDATE employees SET emp_no = 3 WHERE emp_no = 1; füllt daher das BI_emp_no Feld mit 1. Wenn Sie CDC Aktualisierungen für Amazon S3 S3-Ziele schreiben, können Sie anhand der BI_emp_no Spalte erkennen, welche ursprüngliche Zeile aktualisiert wurde.

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

Weitere Informationen zur Verwendung der add-before-image-columns-Regelaktion finden Sie unter Transformationsregeln und Aktionen.

Einschränkungen bei der Verwendung von Apache Kafka als Ziel für AWS Database Migration Service

Bei der Verwendung von Apache Kafka als Ziel gelten die folgenden Einschränkungen:

Verwenden der Objektzuweisung zum Migrieren von Daten zu einem Kafka-Thema

AWS DMS verwendet Regeln für die Tabellenzuordnung, um Daten aus der Quelle dem Kafka-Zielthema zuzuordnen. Um Daten einem Zielthema zuzuweisen, verwenden Sie eine Art von Tabellenzuweisungsregel, die als Objektzuweisung bezeichnet wird. Mit der Objektzuweisung legen Sie fest, wie Datensätze in der Quelle den in einem Kafka-Thema veröffentlichten Datensätzen zugewiesen werden.

Kafka-Themen verfügen bis auf einen Partitionsschlüssel über keine voreingestellte Struktur.

Um eine Objektzuweisungsregel zu erstellen, legen Sie rule-type als object-mapping fest. Diese Regel gibt an, welchen Objektzuweisungstyp Sie verwenden möchten.

Die Struktur für die Regel lautet wie folgt.

{
    "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 unterstützt derzeit map-record-to-record und map-record-to-document als einzig gültige Werte für den rule-action Parameter. Diese Einstellungen wirken sich auf Werte aus, die nicht in der exclude-columns-Attributliste ausgeschlossen sind. Die map-record-to-document Werte map-record-to-record und geben an, wie diese Datensätze standardmäßig AWS DMS behandelt werden. Diese Werte wirken sich in keiner Weise auf die Attributzuweisungen aus.

Verwenden Sie map-record-to-record beim Migrieren aus einer relationalen Datenbank zu einem Kafka-Thema. Dieser Regeltyp verwendet den taskResourceId.schemaName.tableName-Wert aus der relationalen Datenbank als Partitionsschlüssel in dem Kafka-Thema und erstellt ein Attribut für jede Spalte in der Quelldatenbank.

Beachten Sie bei Verwendung von map-record-to-record Folgendes:

Eine Möglichkeit, map-record-to-record zu verstehen, besteht darin, sich die praktische Anwendung zu veranschaulichen. In diesem Beispiel wird davon ausgegangen, dass Sie mit einer Tabellenzeile einer relationalen Datenbank beginnen, die die folgende Struktur aufweist und die folgenden Daten enthält.

Um diese Informationen von einem Schema mit dem Namen Test zu einem Kafka-Thema zu migrieren, erstellen Sie Regeln, um die Daten dem Zielthema zuzuweisen. Die folgende Regel veranschaulicht die Zuweisung.

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

Nachfolgend wird das resultierende Datensatzformat bei Verwendung unserer Beispieldaten in dem Kafka-Zielthema für ein bestimmtes Kafka-Thema und einen bestimmten Partitionsschlüssel (in diesem Fall taskResourceId.schemaName.tableName) illustriert.

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

Umstrukturieren von Daten mit Attributzuweisung

Sie können die Daten umstrukturieren, während Sie sie mittels einer Attributzuweisung zu einem Kafka-Thema migrieren. So möchten Sie zum Beispiel vielleicht mehrere Felder in der Quelle in einem einzigen Feld im Ziel vereinen. Die folgenden Attributzuordnung veranschaulicht, wie die Daten umstrukturiert werden.

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

Um einen konstanten Wert für partition-key festzulegen, geben Sie einen partition-key-Wert an. So könnten Sie auf diese Weise beispielsweise erzwingen, dass alle Daten in einer einzigen Partition gespeichert werden. Die folgende Darstellung veranschaulicht dieses Konzept.

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

Replikation für mehrere Themen mithilfe der Objektzuweisung

Standardmäßig migrieren AWS DMS Aufgaben alle Quelldaten zu einem der folgenden Kafka-Themen:

Bei AWS DMS Engine-Versionen 3.4.6 und höher können Sie das kafka-target-topic Attribut verwenden, um jede migrierte Quelltabelle einem separaten Thema zuzuordnen. Mit den folgenden Objektzuweisungsregeln werden beispielsweise die Quelltabellen Customer und Address zu den Kafka-Themen customer_topic bzw. address_topic migriert. AWS DMS Migriert gleichzeitig alle anderen Quelltabellen, einschließlich der Bills Tabelle im Test Schema, zu dem im Zielendpunkt angegebenen Thema.

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

Mithilfe der Kafka-Replikation für mehrere Themen können Sie Quelltabellen mit einer einzigen Replikationsaufgabe gruppieren und zu separaten Kafka-Themen migrieren.

Nachrichtenformat für Apache Kafka

Die JSON Ausgabe ist einfach eine Liste von Schlüssel-Wert-Paaren.

Das folgende JSON Nachrichtenbeispiel veranschaulicht einen Datentyp „Nachricht“ mit allen zusätzlichen Metadaten.

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

Das folgende JSON Nachrichtenbeispiel veranschaulicht eine Meldung vom Typ Steuerelement.

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