Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Usare Apache Kafka come obiettivo per AWS Database Migration Service

È possibile utilizzarlo AWS DMS per migrare i dati in un cluster Apache Kafka. Apache Kafka è una piattaforma di streaming distribuita. È possibile utilizzare Apache Kafka per l'inserimento e l'elaborazione dei dati di streaming in tempo reale.

AWS offre anche Amazon Managed Streaming for Apache Kafka (MSKAmazon) da utilizzare come destinazione. AWS DMS Amazon MSK è un servizio di streaming Apache Kafka completamente gestito che semplifica l'implementazione e la gestione delle istanze Apache Kafka. Funziona con le versioni open source di Apache Kafka e accedi alle MSK istanze Amazon come AWS DMS destinazioni esattamente come qualsiasi istanza di Apache Kafka. Per ulteriori informazioni, consulta What is AmazonMSK? nella Amazon Managed Streaming for Apache Kafka Developer Guide.

Un cluster Kafka archivia i flussi di record in categorie denominate argomenti che sono suddivisi in partizioni. Le partizioni sono sequenze di record di dati (messaggi) identificate in modo univoco in un argomento. Le partizioni possono essere distribuite tra più broker in un cluster per consentire l'elaborazione parallela dei record dell'argomento. Per ulteriori informazioni su argomenti e partizioni e la loro distribuzione in Apache Kafka, consulta Argomenti e registri e Distribuzione.

Il tuo cluster Kafka può essere un'MSKistanza Amazon, un cluster in esecuzione su un'EC2istanza Amazon o un cluster locale. Un'MSKistanza Amazon o un cluster su un'EC2istanza Amazon può trovarsi nella stessa istanza VPC o in un'altra. Se il cluster è on-premise, è possibile utilizzare il server dei nomi on-premise dell'istanza di replica per risolvere il nome host del cluster. Per informazioni sulla configurazione di un server dei nomi per l'istanza di replica, consulta Utilizzo del server dei nomi in locale. Per ulteriori informazioni sulla configurazione di una rete, consulta Configurazione di una rete per un'istanza di replica.

Quando utilizzi un MSK cluster Amazon, assicurati che il relativo gruppo di sicurezza consenta l'accesso dalla tua istanza di replica. Per informazioni sulla modifica del gruppo di sicurezza per un MSK cluster Amazon, consulta Modifica del gruppo di sicurezza di un MSK cluster Amazon.

AWS Database Migration Service pubblica record su un argomento di Kafka utilizzando. JSON Durante la conversione, AWS DMS serializza ogni record dal database di origine in una coppia attributo-valore in formato. JSON

È possibile usare la mappatura degli oggetti per migrare i dati da qualsiasi origine dati supportata a un cluster Kafka di destinazione. Con la mappatura degli oggetti, determini il modo in cui strutturare i record di dati nell'argomento di destinazione. Puoi inoltre definire una chiave di partizionamento per ogni tabella che viene utilizzata da Kafka per raggruppare i dati nelle sue partizioni.

Attualmente, AWS DMS supporta un singolo argomento per attività. Per una singola attività con più tabelle, tutti i messaggi vengono indirizzati a un unico argomento. Ogni messaggio include una sezione di metadati che identifica lo schema e la tabella di destinazione. AWS DMS le versioni 3.4.6 e successive supportano la replica multitopica utilizzando la mappatura degli oggetti. Per ulteriori informazioni, consulta Replica di più argomenti con la mappatura degli oggetti.

Quando si utilizzano le impostazioni delle attività ParallelApply*, l'impostazione di partition-key-type predefinita è la primary-key della tabella, non schema-name.table-name.

Connessione a Kafka tramite Transport Layer Security () TLS

Un cluster Kafka accetta connessioni sicure utilizzando Transport Layer Security (). TLS ConDMS, è possibile utilizzare una qualsiasi delle seguenti tre opzioni di protocollo di sicurezza per proteggere una connessione endpoint Kafka.

Utilizzo della SSL crittografia con Amazon MSK o un cluster Kafka autogestito

Puoi utilizzare SSL la crittografia per proteggere una connessione endpoint ad Amazon MSK o a un cluster Kafka autogestito. Quando utilizzi il metodo di autenticazione SSL crittografata, i client convalidano l'identità di un server tramite il certificato del server. Quindi viene stabilita una connessione crittografata tra server e client.

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

Utilizzo SSL dell'autenticazione

Puoi utilizzare l'SSLautenticazione per proteggere una connessione endpoint ad Amazon MSK o a un cluster Kafka autogestito.

Per abilitare l'autenticazione e la crittografia dei client utilizzando SSL l'autenticazione per connettersi ad AmazonMSK, procedi come segue:

Utilizzo SASL SSL dell'autenticazione per connettersi ad Amazon MSK

Il metodo Simple Authentication and Security Layer (SASL) utilizza un nome utente e una password per convalidare l'identità di un client e stabilisce una connessione crittografata tra server e client.

Per utilizzarliSASL, devi prima creare un nome utente e una password sicuri quando configuri il tuo MSK cluster Amazon. Per una descrizione di come configurare un nome utente e una password sicuri per un MSK cluster Amazon, consulta ConfigurazioneSASL/SCRAMautenticazione per un MSK cluster Amazon nella Amazon Managed Streaming for Apache Kafka Developer Guide.

Quindi, quando crei l'endpoint di destinazione Kafka, imposta l'impostazione dell'endpoint del protocollo di sicurezza (SecurityProtocol) utilizzando l'opzione sasl-ssl. Imposti anche le opzioni SaslUsername e SaslPassword. Assicurati che siano coerenti con il nome utente e la password sicuri che hai creato quando hai configurato per la prima volta il tuo MSK cluster Amazon, come mostrato nell'JSONesempio seguente.

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

Utilizzo di un'immagine precedente per visualizzare i valori originali delle CDC righe per Apache Kafka come destinazione

Quando si scrivono CDC aggiornamenti su una destinazione di streaming di dati come Kafka, è possibile visualizzare i valori originali di una riga del database di origine prima di modificarli mediante un aggiornamento. Per rendere possibile ciò, AWS DMS compila un'immagine precedente degli eventi di aggiornamento sulla base dei dati forniti dal motore del database di origine.

Diversi motori di database di origine forniscono diverse quantità di informazioni per un'immagine precedente:

Per consentire prima dell'imaging di aggiungere valori originali dal database di origine all'output AWS DMS , utilizzare l'impostazione dell'attività BeforeImageSettings o il parametro add-before-image-columns. Questo parametro applica una regola di trasformazione della colonna.

BeforeImageSettingsaggiunge un nuovo JSON attributo a ogni operazione di aggiornamento con valori raccolti dal sistema di database di origine, come illustrato di seguito.

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

Per le opzioni BeforeImageSettings, si applica quanto segue:

Utilizzo di una regola di trasformazione dell'immagine precedente

In alternativa alle impostazioni delle attività, è possibile utilizzare il parametro add-before-image-columns, che applica una regola di trasformazione delle colonne. Con questo parametro, è possibile abilitare la funzione Before Imaging durante lo streaming di dati CDC su destinazioni come Kafka.

Utilizzando add-before-image-columns in una una regola di trasformazione, è possibile applicare un controllo più dettagliato dei risultati dell'immagine precedente. Le regole di trasformazione consentono di utilizzare un localizzatore di oggetti che consente di controllare le tabelle selezionate per la regola. Inoltre, è possibile concatenare le regole di trasformazione, consentendo l'applicazione di regole diverse a tabelle diverse. È quindi possibile manipolare le colonne prodotte utilizzando altre regole.

Un tipo di regola transformation regola con il parametro add-before-image-columns per una colonna deve fornire una sezione before-image-def. Di seguito viene riportato un esempio.

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

Il valore di column-prefix viene anteposto a un nome di colonna e il valore predefinito di column-prefix è BI_. Il valore di column-suffix viene aggiunto al nome della colonna e il valore predefinito è vuoto. Non impostare sia column-prefix sia column-suffix sull'opzione per svuotare le stringhe.

Scegliere un valore per column-filter. Per aggiungere solo colonne che fanno parte delle chiavi primarie della tabella, scegliere pk-only . Scegliete non-lob di aggiungere solo colonne non tipograficheLOB. Oppure scegliere all per aggiungere qualsiasi colonna con un valore immagine precedente.

Esempio di una regola di trasformazione dell'immagine precedente

La regola di trasformazione nell'esempio seguente aggiunge una nuova colonna chiamata BI_emp_no nella destinazione. Quindi una dichiarazione come UPDATE employees SET emp_no = 3 WHERE emp_no = 1; popola il campo BI_emp_no con 1. Quando scrivi CDC aggiornamenti alle destinazioni Amazon S3, la BI_emp_no colonna consente di determinare quale riga originale è stata aggiornata.

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

Per informazioni sull'utilizzo dell'operazione della regola add-before-image-columns, consulta Operazioni e regole di trasformazione.

Limitazioni nell'utilizzo di Apache Kafka come destinazione per AWS Database Migration Service

Utilizzando Apache Kafka come destinazione valgono le seguenti limitazioni:

Utilizzo della mappatura degli oggetti per la migrazione dei dati in un argomento Kafka

AWS DMS utilizza regole di mappatura delle tabelle per mappare i dati dall'argomento Kafka di origine a quello di destinazione. Per mappare i dati a un argomento di destinazione, è necessario utilizzare una regola di mappatura delle tabelle denominata mappatura degli oggetti. La mappatura degli oggetti consente di definire il modo in cui i record di dati nell'origine vengono mappati ai record di dati pubblicati nell'argomento Kafka.

Gli argomenti Kafka non dispongono di una struttura preimpostata oltre a una chiave di partizione.

Per creare una regola di mappatura degli oggetti, è necessario impostare il parametro rule-type su object-mapping. Questa regola specifica il tipo di mappatura degli oggetti da utilizzare.

Di seguito è riportata la struttura per la regola.

{
    "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 attualmente supporta map-record-to-record e è map-record-to-document l'unico valore valido per il parametro. rule-action Queste impostazioni influiscono sui valori che non sono esclusi come parte dell'elenco degli attributi exclude-columns. I map-record-to-document valori map-record-to-record and specificano come AWS DMS gestisce questi record per impostazione predefinita. Questi valori non influiscono in alcun modo sulle mappature degli attributi.

Nel caso di migrazione da un database relazionale a un argomento Kafka utilizza map-record-to-record. Questo tipo di regola utilizza il valore taskResourceId.schemaName.tableName dal database relazionale come chiave di partizione nell'argomento Kafka e crea un attributo per ogni colonna nel database di origine.

Quando utilizzi map-record-to-record, tieni presente quanto segue:

Per comprendere il funzionamento di map-record-to-record, è opportuno esaminarne il comportamento in azione. Per questo esempio, supponiamo che tu stia iniziando con una riga di tabella del database relazionale con la struttura e i dati seguenti.

Per eseguire la migrazione di queste informazioni da uno schema denominato Test verso un argomento Kafka, crea le regole per mappare i dati sull'argomento di destinazione. La regola seguente illustra la mappatura.

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

Dato un argomento Kafka e una chiave di partizione (in questo caso, taskResourceId.schemaName.tableName), il seguente esempio illustra il formato di record risultante per l'argomento di destinazione Kafka utilizzando i dati di esempio:

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

Ristrutturazione dei dati con la mappatura degli attributi

Utilizzando una mappa degli attributi puoi modificare la struttura dei dati mentre li stai migrando verso un argomento Kafka. Ad esempio, potresti voler combinare più campi nell'origine in un unico campo nella destinazione. La seguente mappa degli attributi illustra come ristrutturare i dati.

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

Per impostare un valore costante per partition-key, specificare un valore partition-key. Ad esempio, potresti eseguire questa operazione per forzare la memorizzazione di tutti i dati in una singola partizione. Questo approccio viene illustrato nella mappatura seguente.

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

Replica di più argomenti con la mappatura degli oggetti

Per impostazione predefinita, AWS DMS le attività migrano tutti i dati di origine su uno degli argomenti di Kafka seguenti:

Con le versioni AWS DMS del motore 3.4.6 e successive, è possibile utilizzare l'kafka-target-topicattributo per mappare ogni tabella di origine migrata su un argomento separato. Ad esempio, le regole di mappatura degli oggetti riportate di seguito migrano le tabelle di origine Customer e Address agli argomenti Kafka customer_topic e address_topic rispettivamente. Allo stesso tempo, AWS DMS migra tutte le altre tabelle di origine, inclusa la Bills tabella nello Test schema, all'argomento specificato nell'endpoint di destinazione.

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

Utilizzando la replica di più argomenti Kafka, è possibile raggruppare e migrare le tabelle di origine su argomenti Kafka separati utilizzando un'unica attività di replica.

Formato dei messaggi per Apache Kafka

L'JSONoutput è semplicemente un elenco di coppie chiave-valore.

Il seguente esempio di JSON messaggio illustra un messaggio di tipo di dati con tutti i metadati aggiuntivi.

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

Il seguente esempio di JSON messaggio illustra un messaggio di tipo di controllo.

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