Utilizzo di Amazon Kinesis Data Streams come destinazione per AWS Database Migration Service

Puoi utilizzarlo AWS DMS per migrare i dati verso un flusso di dati Amazon Kinesis. I flussi di dati Amazon Kinesis fanno parte del servizio Flusso di dati Amazon Kinesis. Puoi utilizzare i flussi di dati Kinesis per raccogliere ed elaborare flussi di grandi dimensioni di record di dati in tempo reale.

Un flusso di dati Kinesis è costituito da partizioni. Gli shard sono sequenze identificate in modo univoco di record di dati in un flusso. Per ulteriori informazioni sulle partizioni in Flusso di dati Amazon Kinesis, consulta Shard nella Guida per gli sviluppatori di Flusso di dati Amazon Kinesis.

AWS Database Migration Service pubblica i record in un flusso di dati Kinesis utilizzando JSON. Durante la conversione, AWS DMS serializza ogni record dal database di origine in una coppia attributo-valore in formato JSON o in un formato di messaggio JSON_UNFORMATTED. Un formato di messaggio JSON_UNFORMATTED è una stringa JSON a riga singola con nuovo delimitatore di riga. Consente ad Amazon Data Firehose di distribuire dati Kinesis a una destinazione Amazon S3 e quindi di interrogarli utilizzando vari motori di query tra cui Amazon Athena.

È possibile utilizzare la mappatura degli oggetti per migrare i dati da qualsiasi origine dati supportata a un flusso di destinazione. Con la mappatura degli oggetti, determini il modo in cui strutturare i record di dati nel flusso. Puoi inoltre definire una chiave di partizione per ogni tabella che viene utilizzata dai flussi di dati Kinesis per raggruppare i dati nelle partizioni.

Quando AWS DMS crea tabelle su un endpoint di destinazione Kinesis Data Streams, crea tante tabelle quante sono nell'endpoint del database di origine. AWS DMS imposta anche diversi valori dei parametri Kinesis Data Streams. Il costo per la creazione della tabella dipende dalla quantità di dati e dal numero di tabelle da migrare.

Nota

L'opzione SSL Mode sulla AWS DMS console o sull'API non si applica ad alcuni servizi di streaming di dati e NoSQL come Kinesis e DynamoDB. Sono sicuri per impostazione predefinita, quindi AWS DMS mostra che l'impostazione della modalità SSL è uguale a none (modalità SSL = Nessuno). Per utilizzare SSL non è necessario eseguire alcuna configurazione aggiuntiva per l'endpoint. Ad esempio, l'utilizzo di Kinesis come endpoint di destinazione è sicuro per impostazione predefinita. Tutte le chiamate API a Kinesis utilizzano SSL, quindi non è necessaria un'opzione SSL aggiuntiva nell'endpoint. AWS DMS È possibile inserire dati e recuperarli in modo sicuro tramite gli endpoint SSL utilizzando il protocollo HTTPS, usato da AWS DMS per impostazione predefinita per la connessione a un flusso di dati Kinesis.

Impostazioni degli endpoint del flusso di dati Kinesis

Quando utilizzi gli endpoint target di Kinesis Data Streams, puoi ottenere i dettagli delle transazioni e del controllo KinesisSettings utilizzando l'opzione nell'API. AWS DMS

Puoi configurare le impostazioni di connessione nei modi seguenti:

• Nella AWS DMS console, utilizzando le impostazioni degli endpoint.

• Nella CLI, utilizzando l'kinesis-settingsopzione del CreateEndpointcomando.

Nella CLI utilizza i seguenti parametri di richiesta dell'opzione kinesis-settings:

Nota

Il supporto per l'impostazione dell'endpoint IncludeNullAndEmpty è disponibile in AWS DMS 3.4.1 e versioni successive. Tuttavia, il supporto per le altre impostazioni degli endpoint seguenti per i target Kinesis Data Streams è disponibile in. AWS DMS

• MessageFormat: il formato di output per i record creati nell'endpoint. Il formato del messaggio è JSON (predefinito) o JSON_UNFORMATTED (una singola riga senza tabulazione).

• IncludeControlDetails: mostra informazioni dettagliate sul controllo per la definizione di tabella, la definizione di colonna e le modifiche di tabelle e colonne nell'output dei messaggi Kinesis. Il valore predefinito è false.

• IncludeNullAndEmpty: include le colonne vuote e NULL nella destinazione. Il valore predefinito è false.

• IncludePartitionValue: mostra il valore della partizione all'interno dell'output dei messaggi di Kinesis, a meno che il tipo della partizione non sia schema-table-type. Il valore predefinito è false.

• IncludeTableAlterOperations: include tutte le operazioni DDL (Data Definition Language) che modificano i dati di controllo della tabella, ad esempio rename-table, drop-table, add-column, drop-column e rename-column. Il valore predefinito è false.

• IncludeTransactionDetails: fornisce informazioni dettagliate sulle transazioni dal database di origine. Tali informazioni includono un timestamp di commit, una posizione nel log e valori per transaction_id, previous_transaction_id e transaction_record_id (l'offset del record all'interno di una transazione). Il valore predefinito è false.

• PartitionIncludeSchemaTable: aggiunge ai nomi di schemi e tabelle il prefisso con i valori di partizione, quando il tipo di partizione è primary-key-type. In questo modo si aumenta la distribuzione dei dati tra gli shard di Kinesis. Ad esempio, si supponga che uno schema SysBench includa migliaia di tabelle e che ogni tabella faccia riferimento solo a un intervallo limitato di valori della chiave primaria. In questo caso, la stessa chiave primaria viene inviata da migliaia di tabelle allo stesso shard, causando un rallentamento. Il valore predefinito è false.

L'esempio seguente mostra l'opzione kinesis-settings in uso con un comando create-endpoint di esempio emesso utilizzando la AWS CLI.

aws dms create-endpoint --endpoint-identifier=$target_name --engine-name kinesis --endpoint-type target
--region us-east-1 --kinesis-settings ServiceAccessRoleArn=arn:aws:iam::333333333333:role/dms-kinesis-role,
StreamArn=arn:aws:kinesis:us-east-1:333333333333:stream/dms-kinesis-target-doc,MessageFormat=json-unformatted,
IncludeControlDetails=true,IncludeTransactionDetails=true,IncludePartitionValue=true,PartitionIncludeSchemaTable=true,
IncludeTableAlterOperations=true

Impostazioni attività a pieno carico multithread

Per contribuire ad aumentare la velocità di trasferimento, AWS DMS supporta un caricamento completo multithread su un'istanza di destinazione Kinesis Data Streams. DMS supporta questo multithreading con impostazioni delle attività che includono le seguenti:

• MaxFullLoadSubTasks: imposta questa opzione per indicare il numero massimo di tabelle da caricare in parallelo. DMS consente di caricare ogni tabella nella corrispondente tabella di destinazione Kinesis utilizzando un'attività secondaria dedicata. Il valore predefinito è 8; il valore il massimo è 49.

• ParallelLoadThreads— Utilizzate questa opzione per specificare il numero di thread da utilizzare per caricare ogni tabella nella relativa tabella di destinazione Kinesis. AWS DMS Il valore massimo per una destinazione del flusso di dati Kinesis è 32. Puoi chiedere che questo limite massimo venga aumentato.

• ParallelLoadBufferSize: utilizza questa opzione per specificare il numero massimo di record da archiviare nel buffer utilizzato dai thread di caricamento parallelo per caricare i dati nella destinazione Kinesis. Il valore predefinito è 50. Il valore massimo è 1.000. Utilizzare questo parametro con ParallelLoadThreads; ParallelLoadBufferSize è valido solo quando è presente più di un thread.

• ParallelLoadQueuesPerThread: utilizza questa opzione per specificare il numero di code a cui accede ogni thread simultaneo per eliminare i record di dati dalle code e generare un carico batch per la destinazione. Il valore di default è 1. Tuttavia, per le destinazioni Kinesis di varie dimensioni del payload, l'intervallo valido è compreso tra 5 e 512 code per thread.

Impostazioni attività di carico CDC multithread

È possibile migliorare le prestazioni dell'acquisizione dei dati di modifica (CDC) per gli endpoint di destinazione in streaming dei dati in tempo reale come Kinesis utilizzando le impostazioni delle attività per modificare il comportamento della chiamata API PutRecords. A tale scopo, è possibile specificare il numero di thread simultanei, di code per thread e di record da memorizzare in un buffer utilizzando le impostazioni delle attività ParallelApply*. Ad esempio, si supponga di voler eseguire un carico CDC e applicare 128 thread in parallelo. Si desidera inoltre accedere a 64 code per thread, con 50 record memorizzati per buffer.

Per promuovere le prestazioni del CDC, AWS DMS supporta le seguenti impostazioni delle attività:

• ParallelApplyThreads— specifica il numero di thread simultanei che vengono AWS DMS utilizzati durante un caricamento CDC per inviare i record di dati a un endpoint di destinazione Kinesis. Il valore predefinito è zero (0) e il valore massimo è 32.

• ParallelApplyBufferSize: specifica il numero massimo di record da archiviare in ogni coda di buffer per eseguire il push dei thread simultanei a un endpoint di destinazione Kinesis durante un carico CDC. Il valore predefinito è 100 e il valore massimo è 1.000. Utilizzare questa opzione quando ParallelApplyThreads specifica più di un thread.

• ParallelApplyQueuesPerThread: specifica il numero di code a cui ogni thread accede per eliminare i record di dati dalle code e generare un carico batch per un endpoint Kinesis durante CDC. Il valore predefinito è 1 e il valore massimo è 512.

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.

Utilizzo di un'immagine precedente per visualizzare i valori originali delle righe CDC per un flusso di dati Kinesis come destinazione

Quando si scrivono aggiornamenti CDC su una destinazione di streaming dati come Kinesis, è 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 di database di origine.

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

• Oracle fornisce aggiornamenti alle colonne solo se cambiano.

• PostgreSQL fornisce solo i dati per le colonne che fanno parte della chiave primaria (modificata o meno). Per fornire dati per tutte le colonne (modificate o meno), devi impostare su REPLICA_IDENTITY su FULL invece di DEFAULT. Tieni presente che occorre scegliere attentamente l'impostazione REPLICA_IDENTITY per ogni tabella. Se si imposta REPLICA_IDENTITY su FULL, tutti i valori delle colonne vengono scritti continuamente nel WAL (Write-Ahead Logging). Ciò può causare problemi di prestazioni o di risorse con le tabelle che vengono aggiornate frequentemente.

• MySQL generalmente fornisce dati per tutte le colonne ad eccezione dei tipi di dati BLOB e CLOB (modificati o meno).

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)
}
        

Nota

Si applica solo BeforeImageSettings alle AWS DMS attività che contengono un componente CDC, come le attività a pieno carico e le attività CDC (che migrano i dati esistenti e replicano le modifiche in corso) o alle attività solo CDC (che replicano solo le modifiche ai dati). Non applicare BeforeImageSettings alle attività a pieno carico.

Per le opzioni BeforeImageSettings, si applica quanto segue:

• Impostare l'opzione EnableBeforeImage su true da abilitare prima dell'imaging. Il valore predefinito è false.

• Utilizzare l'opzione FieldName per assegnare un nome al nuovo attributo JSON. Quando EnableBeforeImage è true, FieldName è richiesto e non può essere vuoto.

• L'opzione ColumnFilter specifica una colonna da aggiungere utilizzando l'imaging precedente. Per aggiungere solo colonne che fanno parte delle chiavi primarie della tabella, utilizzare il valore predefinito, pk-only. Per aggiungere qualsiasi colonna con un valore immagine prima, utilizzare all. Tieni presente che l'immagine precedente non contiene colonne con tipi di dati LOB, come CLOB o BLOB.

  "BeforeImageSettings": {
      "EnableBeforeImage": true,
      "FieldName": "before-image",
      "ColumnFilter": "pk-only"
    }
  

Nota

Le destinazioni Amazon S3 non supportano BeforeImageSettings. Per le destinazioni S3, utilizzare solo la regola di trasformazione add-before-image-columns da eseguire prima dell'imaging durante CDC.

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 CDC su destinazioni di flusso di dati come Kinesis.

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.

Nota

Non utilizzare il parametro add-before-image-columns insieme all'impostazione dell'attività BeforeImageSettings all'interno della stessa attività. Utilizzare invece il parametro o l'impostazione, ma non entrambi, per una singola attività.

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.

Prerequisiti per l'utilizzo di un flusso di dati Kinesis come destinazione per AWS Database Migration Service

Ruolo IAM per l'utilizzo di un flusso di dati Kinesis come destinazione per AWS Database Migration Service

Prima di configurare un flusso di dati Kinesis come destinazione per AWS DMS, assicurati di creare un ruolo IAM. Questo ruolo deve consentire di AWS DMS assumere e concedere l'accesso ai flussi di dati Kinesis in cui viene effettuata la migrazione. Nella seguente policy IAM viene mostrato il set minimo di autorizzazioni di accesso.

{
   "Version": "2012-10-17",
   "Statement": [
   {
     "Sid": "1",
     "Effect": "Allow",
     "Principal": {
        "Service": "dms.amazonaws.com"
     },
   "Action": "sts:AssumeRole"
   }
]
} 

        

Il ruolo utilizzato per la migrazione a un flusso di dati Kinesis deve disporre delle seguenti autorizzazioni.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:region:accountID:stream/streamName"
        }
    ]
}

Accesso a un flusso di dati Kinesis come destinazione per AWS Database Migration Service

Nella AWS DMS versione 3.4.7 e successive, per connetterti a un endpoint Kinesis, devi eseguire una delle seguenti operazioni:

• Configura DMS per utilizzare gli endpoint VPC. Per informazioni sulla configurazione di DMS per utilizzare gli endpoint VPC, consulta Configurazione degli endpoint VPC come endpoint di origine e di destinazione AWS DMS.

• Configura DMS per utilizzare routing pubblici, ovvero per rendere pubblica l'istanza di replica. Per ulteriori informazioni sulle istanze di replica pubbliche, consulta Istanze di replica pubbliche e private.

Limitazioni nell'utilizzo di Kinesis Data Streams come destinazione per AWS Database Migration Service

Le seguenti limitazioni si applicano quando si utilizza il flusso di dati Kinesis come destinazione:

• AWS DMS pubblica ogni aggiornamento di un singolo record nel database di origine come un unico record di dati in un determinato flusso di dati Kinesis indipendentemente dalle transazioni. Tuttavia, è possibile includere i dettagli della transazione per ogni record di dati utilizzando i parametri pertinenti dell'API KinesisSettings.

• La modalità LOB completa non è supportata.

• La dimensione massima del LOB supportata è 1 MB.

• I flussi di dati Kinesis non supportano la deduplicazione. Le applicazioni che utilizzano i dati provenienti da un flusso devono gestire i record duplicati. Per ulteriori informazioni, consulta Handling duplicate records nella Guida per gli sviluppatori di Flusso di dati Amazon Kinesis.

• AWS DMS supporta i seguenti due moduli per le chiavi di partizione:

  • SchemaName.TableName: una combinazione del nome dello schema e quello della tabella.

  • ${AttributeName}: il valore di uno dei campi nel formato JSON o la chiave primaria della tabella del database di origine.

• Per informazioni sulla crittografia dei dati a riposo all'interno del flusso di dati Kinesis, consulta Data protection in Kinesis Data Streams nella Guida per gli sviluppatori di AWS Key Management Service .

• BatchApply non è supportato per un endpoint Kinesis. L'utilizzo dell'applicazione in batch, ad esempio l'impostazione dell'attività dei metadati di destinazione BatchApplyEnabled, per una destinazione Kinesis potrebbe causare la perdita di dati.

• Le destinazioni Kinesis sono supportate solo per un flusso di dati Kinesis nello stesso AWS account e nello stesso dell'istanza di replica. Regione AWS

• Durante la migrazione da una fonte MySQL, i dati non includono BeforeImage i tipi di dati CLOB e BLOB. Per ulteriori informazioni, consulta Utilizzo di un'immagine precedente per visualizzare i valori originali delle righe CDC per un flusso di dati Kinesis come destinazione.

• AWS DMS non supporta la migrazione di valori di tipi di BigInt dati con più di 16 cifre. Per aggirare questa limitazione puoi utilizzare la seguente regola di trasformazione per convertire la colonna BigInt in una stringa. Per ulteriori informazioni sulle regole di trasformazione, consulta Operazioni e regole di trasformazione.

  {
      "rule-type": "transformation",
      "rule-id": "id",
      "rule-name": "name",
      "rule-target": "column",
      "object-locator": {
          "schema-name": "valid object-mapping rule action",
          "table-name": "",
          "column-name": ""
      },
      "rule-action": "change-data-type",
      "data-type": {
          "type": "string",
          "length": 20
      }
  }

Utilizzo della mappatura degli oggetti per la migrazione dei dati a un flusso di dati Kinesis

AWS DMS utilizza regole di mappatura delle tabelle per mappare i dati dal flusso di dati Kinesis di origine a quello di destinazione. Per mappare i dati a un flusso 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 nel flusso di dati Kinesis.

I flussi di dati Kinesis non dispongono di una struttura preimpostata oltre a una chiave di partizione. In una regola di mapping degli oggetti, i valori possibili di una partition-key-type per i record di dati sono schema-table, transaction-id, primary-key, constant e attribute-name.

Per creare una regola di mappatura degli oggetti, è necessario specificare il parametro rule-type come 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.

Utilizza map-record-to-record per la migrazione da un database relazionale a un flusso di dati Kinesis. Questo tipo di regola utilizza il valore taskResourceId.schemaName.tableName dal database relazionale come chiave di partizione nel flusso di dati Kinesis e crea un attributo per ogni colonna nel database di origine.

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

• Questa impostazione ha effetto solo sulle colonne escluse dall'elenco exclude-columns.

• Per ogni colonna di questo tipo, AWS DMS crea un attributo corrispondente nell'argomento di destinazione.

• AWS DMS crea questo attributo corrispondente indipendentemente dal fatto che la colonna di origine venga utilizzata in una mappatura degli attributi.

Utilizza map-record-to-document per inserire le colonne di origine in un unico documento flat nel flusso di destinazione appropriato utilizzando il nome dell'attributo "_doc". AWS DMS posiziona i dati in un'unica mappa flat sull'origine chiamata "_doc". Questo posizionamento si applica a qualsiasi colonna della tabella di origine non elencata nell'elenco di attributi exclude-columns.

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.

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

Per eseguire la migrazione di queste informazioni da uno schema denominato Test a un flusso di dati Kinesis, crea le regole per mappare i dati sul flusso 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": "DefaultMapToKinesis",
            "rule-action": "map-record-to-record",
            "object-locator": {
                "schema-name": "Test",
                "table-name": "Customers"
            }
        }
    ]
}

Di seguito viene illustrato il formato di record risultante nel flusso di dati Kinesis:

• StreamName: XXX

• PartitionKey: Test.Customers //schmaname.tableName

• Dati: //Il seguente messaggio JSON

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

Supponi tuttavia di utilizzare le stesse regole ma modificando il parametro rule-action su map-record-to-document ed escludendo determinate colonne. 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": "DefaultMapToKinesis",
			"rule-action": "map-record-to-document",
			"object-locator": {
				"schema-name": "Test",
				"table-name": "Customers"
			},
			"mapping-parameters": {
				"exclude-columns": [
					"homeaddress",
					"homephone",
					"workaddress",
					"workphone"
				]
			}
		}
	]
}                

In questo caso, le colonne non elencate nel parametro exclude-columns, FirstName, LastName, StoreId e DateOfBirth sono mappate a _doc. Di seguito viene illustrato il formato di record risultante.

       {
            "data":{
                "_doc":{
                    "FirstName": "Randy",
                    "LastName": "Marsh",
                    "StoreId":  "5",
                    "DateOfBirth": "02/29/1988"
                }
            }
        }
            

Ristrutturazione dei dati con la mappatura degli attributi

Puoi ristrutturare i dati mentre li stai migrando a un flusso di dati Kinesis utilizzando una mappa degli attributi. 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": "TransformToKinesis",
            "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 l'archiviazione di tutti i dati in un singolo shard. 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": "TransformToKinesis",
            "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}"
                    }
                ]
            }
        }
    ]
}

Nota

Il valore partition-key per un record di controllo per una tabella specifica è TaskId.SchemaName.TableName. Il valore partition-key per un record di controllo per un'attività specifica è il TaskId del record. La specifica si un valore partition-key nella mappatura degli oggetti non influisce sul parametro partition-key per un record di controllo.

Formato del messaggio per il flusso di dati Kinesis

L'output JSON è semplicemente un elenco di coppie chiave-valore. Un formato di messaggio JSON_UNFORMATTED è una stringa JSON a riga singola con nuovo delimitatore di riga.

AWS DMS fornisce i seguenti campi riservati per semplificare l'utilizzo dei dati provenienti da Kinesis Data Streams:

RecordType

Il tipo di record può essere relativo ai dati o al controllo. I record di dati rappresentano le righe effettive nell'origine. I record di controllo sono per eventi importanti nel flusso, ad esempio un riavvio dell'attività.

Operazione

Per i record di dati, l'operazione può essere load, insert, update o delete.

Per i record di controllo, l'operazione può essere create-table, rename-table, drop-table, change-columns, add-column, drop-column, rename-column o column-type-change.

SchemaName

Lo schema di origine per il record. Questo campo può essere vuoto per un record di controllo.

TableName

La tabella di origine per il record. Questo campo può essere vuoto per un record di controllo.

Timestamp

Il timestamp relativo al momento della creazione del messaggio JSON. Il campo viene formattato con il formato ISO 8601.