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 (Amazon MSK) 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 istanze Amazon MSK come AWS DMS destinazioni esattamente come qualsiasi istanza di Apache Kafka. Per ulteriori informazioni, consulta Che cos'è Amazon MSK? nella Guida per sviluppatori di Streaming gestito da Amazon per Apache Kafka.

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 cluster Kafka può essere un'istanza Amazon MSK, un cluster in esecuzione su un'istanza Amazon EC2 o un cluster on-premise. Un'istanza Amazon MSK o un cluster su un'istanza Amazon EC2 può trovarsi nello stesso VPC o in uno diverso. 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 cluster Amazon MSK, assicurati che il relativo gruppo di sicurezza consenta l'accesso dall'istanza di replica. Per informazioni sulla modifica del gruppo di sicurezza di un cluster Amazon MSK, consulta Modifica del gruppo di sicurezza di un cluster Amazon MSK.

AWS Database Migration Service pubblica i record relativi a 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 utilizzando Transport Layer Security (TLS)

Il cluster Kafka accetta solo connessioni protette con Transport Layer Security (TLS). Con DMS, è possibile utilizzare una qualsiasi delle seguenti tre opzioni di protocollo di sicurezza per proteggere la connessione degli endpoint Kafka.

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

È possibile utilizzare la crittografia SSL per proteggere la connessione di un endpoint ad Amazon MSK o a un cluster Kafka autogestito. Quando usi il metodo di autenticazione con crittografia SSL, 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 dell'autenticazione SSL

È possibile utilizzare l'autenticazione SSL per proteggere la connessione di un endpoint ad Amazon MSK o a un cluster Kafka autogestito.

Per abilitare l'autenticazione e la crittografia del client utilizzando l'autenticazione SSL per la connessione ad Amazon MSK, procedi come segue:

Utilizzo dell'autenticazione SASL-SSL 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 utilizzare SASL, devi prima creare un nome utente e una password sicuri quando configuri il cluster Amazon MSK. Per una descrizione di come configurare un nome utente e una password sicuri per un cluster Amazon MSK, consulta Configurazione dell'autenticazione SASL/SCRAM per un cluster Amazon MSK nella Guida per sviluppatori di Streaming gestito da Amazon per Apache Kafka.

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 creati durante la configurazione del cluster Amazon MSK, come mostrato nel seguente esempio JSON.

                   
"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 righe CDC per Apache Kafka come destinazione

Quando si scrivono aggiornamenti CDC su una destinazione di streaming dati come Kafka, è possibile visualizzare i valori originali di una riga di database di origine prima di apportare modifiche da 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.

BeforeImageSettings aggiunge un nuovo attributo JSON 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 l'imaging precedente durante il CDC su destinazioni di streaming dati 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 . Scegliere non-lob per aggiungere solo colonne non di tipo LOB. 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 si scrivono aggiornamenti CDC alle destinazioni Amazon S3, la colonna BI_emp_no indica 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'output JSON è semplicemente un elenco di coppie chiave-valore.

Il seguente esempio di messaggio JSON illustra un 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 messaggio JSON illustra un 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"
   }
}