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à.
# Connettori e utility
Amazon EMR fornisce una serie di connettori e utilità per accedere ad altri AWS servizi come fonti di dati. In genere, puoi accedere ai dati in questi servizi dall'interno di un programma. Ad esempio, puoi specificare un flusso Kinesis in una query Hive, uno script Pig o MapReduce un'applicazione e quindi operare su tali dati.
**Topics**
+ [Esportazione, importazione, esecuzione di query e unione di tabelle in DynamoDB con Amazon EMR](EMRforDynamoDB.md)
+ [Kinesis](emr-kinesis.md)
+ [S3 (s3-dist-cp) DistCp](UsingEMR_s3distcp.md)
+ [Pulizia dopo lavori S3 non riusciti DistCp](#s3distcp-cleanup)
# Esportazione, importazione, esecuzione di query e unione di tabelle in DynamoDB con Amazon EMR
**Nota**
Il connettore Amazon EMR-DynamoDB è open source su. GitHub Per ulteriori informazioni, consulta [https://github.com/awslabs/emr-dynamodb-connector](https://github.com/awslabs/emr-dynamodb-connector).
DynamoDB è un servizio di database NoSQL interamente gestito che combina prestazioni elevate e prevedibili con una scalabilità continua. Gli sviluppatori possono creare una tabella di database e accrescerne la richiesta di traffico o lo storage senza limiti. DynamoDB distribuisce automaticamente i dati e il traffico per la tabella su un numero sufficiente di server per gestire la capacità di richieste specificata dal cliente e la quantità di dati archiviati, garantendo al contempo prestazioni rapide e costanti. Utilizzando Amazon EMR e Hive è possibile elaborare in modo rapido ed efficiente grandi quantità di dati, ad esempio i dati archiviati in DynamoDB. Per ulteriori informazioni su DynamoDB, consulta [Guida per gli sviluppatori di Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/).
Apache Hive è un livello software che è possibile utilizzare per eseguire query su cluster MapReduce utilizzando un linguaggio di query semplificato, simile a SQL, denominato HiveQL, e che viene eseguito su architettura Hadoop. Per ulteriori informazioni su Hive e HiveQL, consulta il [Manuale sul linguaggio HiveQL](https://cwiki.apache.org/confluence/display/Hive/LanguageManual). Per ulteriori informazioni su Hive e Amazon EMR, consulta [Apache Hive](emr-hive.md).
È possibile utilizzare Amazon EMR con una versione personalizzata di Hive, che include la connettività a DynamoDB per eseguire operazioni sui dati archiviati in DynamoDB:
+ Caricamento di dati DynamoDB nel file system distribuito Hadoop (HDFS) e loro utilizzo come input in un cluster Amazon EMR.
+ Esecuzione di query in diretta sui dati DynamoDB utilizzando istruzioni di tipo SQL (HiveQL).
+ Unione di dati archiviati in DynamoDB ed esportazione o esecuzione di query sui dati uniti.
+ Esportazione dei dati archiviati in DynamoDB su Amazon S3.
+ Importazione dei dati archiviati in Amazon S3 su DynamoDB.
**Nota**
Il connettore Amazon EMR-DynamoDB non supporta i cluster configurati per utilizzare l'[autenticazione Kerberos](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-kerberos.html).
Potrai eseguire ognuna delle attività seguenti avviando un cluster Amazon EMR, specificando il percorso dei dati in DynamoDB ed eseguendo comandi Hive per gestire i dati in DynamoDB.
Esistono diversi modi per avviare un cluster Amazon EMR: puoi utilizzare la console Amazon EMR, l'interfaccia a riga di comando (CLI) oppure puoi programmare il cluster utilizzando un SDK AWS o l'API Amazon EMR. È anche possibile scegliere se eseguire un cluster Hive interattivamente o da uno script. In questa sezione illustreremo come avviare un cluster Hive interattivo dalla console di Amazon EMR e dalla CLI.
L'utilizzo interattivo di Hive è perfetto per testare le prestazioni delle query e ottimizzare l'applicazione. Dopo aver creato un set di comandi Hive che verrà eseguito periodicamente, valuta la possibilità di creare uno script Hive che Amazon EMR può eseguire senza interventi da parte tua.
**avvertimento**
Le operazioni di lettura o scrittura di Amazon EMR su una tabella DynamoDB vengono conteggiate ai fini della velocità effettiva assegnata stabilita, aumentando potenzialmente la frequenza delle eccezioni alla velocità effettiva assegnata. Per le richieste di grandi dimensioni, Amazon EMR implementa tentativi con backoff esponenziale per gestire il carico di richieste sulla tabella DynamoDB. L'esecuzione contemporanea di processi Amazon EMR con altro traffico potrebbe causare il superamento del livello di velocità effettiva assegnata allocata. Puoi monitorarlo controllando la **ThrottleRequests**metrica in Amazon CloudWatch. Se il carico di richieste è troppo elevato, è possibile riavviare il cluster e impostare [Impostazione della percentuale di lettura](EMR_Hive_Optimizing.md#ReadPercent) o [Impostazione della percentuale di scrittura](EMR_Hive_Optimizing.md#WritePercent) su un valore inferiore per limitare le operazioni Amazon EMR. Per ulteriori informazioni sulle impostazioni di velocità effettiva di DynamoDB, consulta [Velocità effettiva assegnata](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithDDTables.html#ProvisionedThroughput).
Se una tabella è configurata per la [Modalità on demand](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.ReadWriteCapacityMode.html#HowItWorks.OnDemand), è necessario ripristinare la tabella in modalità di provisioning prima di eseguire un'operazione di esportazione o importazione. Le pipeline necessitano di un rapporto di velocità effettiva per calcolare le risorse da utilizzare da una Dynamo. DBtable La modalità on demand rimuove la velocità effettiva di provisioning. Per fornire la capacità di throughput, puoi utilizzare i parametri di Amazon CloudWatch Events per valutare il throughput aggregato utilizzato da una tabella.
**Topics**
+ [Impostazione di una tabella Hive per l'esecuzione dei comandi Hive](EMR_Interactive_Hive.md)
+ [Esempi di comandi Hive per l'esportazione, l'importazione e l'esecuzione di query sui dati in DynamoDB](EMR_Hive_Commands.md)
+ [Ottimizzazione delle prestazioni per le operazioni Amazon EMR in DynamoDB](EMR_Hive_Optimizing.md)
# Impostazione di una tabella Hive per l'esecuzione dei comandi Hive
Apache Hive è un'applicazione di data warehouse che è possibile usare per eseguire query sui dati contenuti nei cluster Amazon EMR utilizzando un linguaggio di tipo SQL. Per ulteriori informazioni su Hive, visita la pagina Web all'indirizzo [http://hive.apache.org/](http://hive.apache.org/).
La procedura seguente presuppone che sia già stato creato un cluster e specificata una coppia di chiavi Amazon EC2. Per ulteriori informazioni su come iniziare a creare cluster, consulta [Nozioni di base su Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs) nella *Guida alla gestione di Amazon EMR*.
## Configura Hive per l'uso MapReduce
Quando usi Hive su Amazon EMR per eseguire query sulle tabelle DynamoDB, possono verificarsi degli errori se Hive utilizza il motore di esecuzione di default, Tez. Per questo motivo, quando crei un cluster con Hive che si integra con DynamoDB come descritto in questa sezione, ti consigliamo di utilizzare una classificazione di configurazione che imposti Hive da utilizzare. MapReduce Per ulteriori informazioni, consulta [Configurazione delle applicazioni](emr-configure-apps.md).
Il seguente frammento mostra la classificazione della configurazione e la proprietà da utilizzare per impostare MapReduce come motore di esecuzione per Hive:
```
[
{
"Classification": "hive-site",
"Properties": {
"hive.execution.engine": "mr"
}
}
]
```
**Per eseguire i comandi Hive in modo interattivo**
1. Connettersi al nodo master. Per maggiori informazioni, consulta [Connessione al nodo master tramite SSH](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html) nella *Guida alla gestione di Amazon EMR*.
1. Nel prompt dei comandi per il nodo master corrente, digitare `hive`.
Viene visualizzato un prompt hive: `hive>`
1. Inserisci un comando Hive che mappi una tabella nell'applicazione Hive ai dati in DynamoDB. Questa tabella agisce come riferimento ai dati archiviati in Amazon DynamoDB: i dati non vengono memorizzati localmente in Hive e le query che utilizzano la tabella sono eseguite sui dati live in DynamoDB. Pertanto, la capacità di lettura o scrittura della tabella viene consumata ogni volta che viene eseguito un comando. Se è prevista l'esecuzione di più comandi Hive sullo stesso set di dati, valutare la possibilità di esportarli.
Di seguito è mostrata la sintassi per la mappatura di una tabella Hive a una tabella DynamoDB.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...");
```
Le tabelle create in Hive da DynamoDB devono essere create come tabelle esterne utilizzando la parola chiave `EXTERNAL`. La differenza tra le tabelle interne ed esterne è che, nel primo caso, i dati vengono eliminati quando viene rimossa la tabella interna. In caso di connessione ad Amazon DynamoDB non si tratta di un comportamento ottimale, quindi sono supportate solo le tabelle esterne.
Ad esempio, il seguente comando Hive crea una tabella denominata *hivetable1* in Hive che fa riferimento alla tabella DynamoDB denominata *dynamodbtable1*. La *tabella* DynamoDB dynamodbtable1 ha uno schema a chiave primaria. hash-and-range L'elemento della chiave hash è `name` (tipo String), l'elemento della chiave di intervallo è `year` (tipo Numeric) e ciascuna voce ha un valore attributo per `holidays` (tipo String Set).
```
CREATE EXTERNAL TABLE hivetable1 (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
```
La riga 1 usa l'istruzione HiveQL `CREATE EXTERNAL TABLE`. Per *hivetable1*, è necessario stabilire una colonna per ogni coppia nome-valore dell'attributo nella tabella DynamoDB e fornire il tipo di dati. Questi valori non prevedono distinzione fra maiuscole e minuscole ed è possibile dare alle colonne qualsiasi nome (eccetto le parole riservate).
La riga 2 usa l'istruzione `STORED BY`. Il valore di `STORED BY` è il nome della classe che gestisce la connessione tra Hive e DynamoDB. Deve essere impostato su `'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'`.
La riga 3 usa l'istruzione `TBLPROPERTIES` per associare "hivetable1" alla tabella e allo schema corretti in DynamoDB. Fornire `TBLPROPERTIES` con i valori per i parametri `dynamodb.table.name` e `dynamodb.column.mapping`. Per questi valori *è prevista* la distinzione tra maiuscole e minuscole.
**Nota**
Tutti i nomi di attributi DynamoDB per la tabella devono avere colonne corrispondenti nella tabella Hive. A seconda della versione di Amazon EMR in uso, se la one-to-one mappatura non esiste, si verificano i seguenti scenari:
In Amazon EMR versione 5.27.0 e successive, il connettore dispone di convalide che garantiscono una mappatura one-to-one tra i nomi degli attributi DynamoDB e le colonne nella tabella Hive. Si verificherà un errore se la mappatura non esiste. one-to-one
In Amazon EMR 5.26.0 e versioni precedenti, la tabella Hive non contiene la coppia nome-valore proveniente da DynamoDB. Se non vengono mappati gli attributi della chiave primaria DynamoDB, Hive genera un errore. Se non vengono mappati gli attributi della chiave non primaria, non vengono generati errori, ma nella tabella Hive non saranno visualizzati i dati. Se i tipi di dati non corrispondono, il valore è null.
A questo punto puoi iniziare a eseguire operazioni Hive su *hivetable1*. Le query eseguite per *hivetable1* vengono eseguite internamente nella tabella DynamoDB *dynamodbtable1* dell'account DynamoDB, consumando unità di lettura o scrittura a ogni esecuzione.
Quando esegui query Hive su una tabella DynamoDB, devi accertarti di aver assegnato una quantità sufficiente di unità di capacità di lettura.
Ad esempio, supponiamo che tu disponga di 100 unità di capacità di lettura assegnate per la tabella DynamoDB. Questo ti consentirà di eseguire 100 operazioni di lettura, o 409.600 byte, al secondo. Se la tabella contiene 20 GB di dati (21.474.836.480 byte) e la tua query Hive esegue una scansione completa della tabella, puoi stimare la durata dell'esecuzione della query:
* 21.474.836.480 / 409.600 = 52.429 secondi = 14,56 ore *
Il solo modo per ridurre il tempo necessario sarebbe modificare le unità di capacità di lettura nella tabella DynamoDB di origine. L'aggiunta di più nodi Amazon EMR non è utile.
Nell'output Hive, la percentuale di completamento viene aggiornata quando terminano uno o più processi del mappatore. Per una tabella DynamoDB di grandi dimensioni con una bassa capacità di lettura assegnata, l'output della percentuale di completamento potrebbe non essere aggiornato per molto tempo; in questo caso, il processo sembrerà essere completo allo 0% per diverse ore. Per uno stato più dettagliato dell'avanzamento del processo, accedi alla console di Amazon EMR, dalla quale potrai visualizzare lo stato delle singole attività del mappatore e statistiche per le letture di dati. Puoi anche accedere all'interfaccia Hadoop sul nodo master e vedere le statistiche di Hadoop. Ti sarà mostrato lo stato della singola attività del mappatore e alcune statistiche di lettura dei dati. Per ulteriori informazioni, consulta i seguenti argomenti:
+ [Interfacce Web ospitate sul nodo master](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html)
+ [Visualizzazione delle interfacce Web di Hadoop](https://docs.aws.amazon.com/emr/latest/ManagementGuide/UsingtheHadoopUserInterface.html)
Per ulteriori informazioni sulle istruzioni HiveQL di esempio per eseguire attività quali l'esportazione o l'importazione di dati da DynamoDB e unione di tabelle, consulta [Esempi di comandi Hive per l'esportazione, l'importazione e l'esecuzione di query sui dati in DynamoDB](EMR_Hive_Commands.md).
**Per annullare una richiesta Hive**
Quando esegui una query Hive, la risposta iniziale del server include il comando per annullare la richiesta. Per annullare la richiesta in qualsiasi momento del processo, utilizza il comando **Kill Command (Comando di chiusura)** dalla risposta del server.
1. Immettere `Ctrl+C` per chiudere il client a riga di comando.
1. Al prompt della shell, immettere il comando **Kill Command (Comando di chiusura)** tratto dalla risposta iniziale del server alla richiesta.
In alternativa, è possibile eseguire il comando seguente dalla riga di comando del nodo master per terminare il processo Hadoop, dove si *job-id* trova l'identificatore del processo Hadoop e può essere recuperato dall'interfaccia utente Hadoop.
```
hadoop job -kill job-id
```
## Tipi di dati per Hive e DynamoDB
La seguente tabella mostra i tipi di dati Hive disponibili, il tipo di DynamoDB predefinito a cui fanno riferimento e i tipi di DynamoDB alternativi a cui possono essere associati.
| Tipo Hive | Tipo di DynamoDB predefinito | Tipo/i di DynamoDB alternativo/i |
| --- | --- | --- |
| stringa | string (S) | |
| bigint o double | number (N) | |
| binary | binary (B) | |
| booleano | boolean (BOOL) | |
| array | list (L) | number set (NS), string set (SS) o binary set (BS) |
| map | elemento | map (M) |
| map | map (M) | |
| | null (NULL) | |
Se desideri scrivere i dati Hive come un tipo corrispondente di DynamoDB alternativo o se i dati di DynamoDB contengono valori di attributi di un tipo di DynamoDB alternativo, puoi specificare la colonna e il tipo di DynamoDB con il parametro `dynamodb.type.mapping`. L'esempio seguente mostra la sintassi per specificare una mappatura di tipi alternativi.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...",
"dynamodb.type.mapping" = "hive_column1_name:dynamodb_attribute1_datatype");
```
Il parametro della mappatura dei tipi è facoltativo e deve essere specificato solo per le colonne che utilizzano tipi alternativi.
Ad esempio, il seguente comando Hive crea una tabella denominata `hivetable2` che fa riferimento alla tabella `dynamodbtable2` di DynamoDB. È simile a `hivetable1`, tranne per il fatto che associa la colonna `col3` al tipo string set (SS).
```
CREATE EXTERNAL TABLE hivetable2 (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable2",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays",
"dynamodb.type.mapping" = "col3:SS");
```
In Hive, `hivetable1` e `hivetable2` sono identici. Tuttavia, quando i dati di tali tabelle vengono scritti nelle tabelle DynamoDB corrispondenti, `dynamodbtable1` contiene elenchi, mentre `dynamodbtable2` contiene set di stringhe.
Se desideri scrivere valori `null` di Hive come attributi del tipo `null` di DynamoDB, puoi farlo con il parametro `dynamodb.null.serialization`. L'esempio seguente mostra la sintassi per specificare la serializzazione `null`.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...",
"dynamodb.null.serialization" = "true");
```
Il parametro di serializzazione null è facoltativo e, se non specificato, è impostato su `false`. Nota che gli attributi `null` di DynamoDB sono letti come valori `null` in Hive indipendentemente dalle impostazioni del parametro. Le raccolte Hive con valori `null` possono essere scritte in DynamoDB solo se il parametro di serializzazione null è specificato come `true`. In caso contrario, si verifica un errore Hive.
Il tipo bigint in Hive è uguale al tipo long in Java, mentre il tipo double di Hive è uguale al tipo double in Java, in termini di precisione. Questo significa che, se disponi di dati numerici memorizzati in DynamoDB con una precisione superiore a quella disponibile nei tipi di dati Hive, l'utilizzo di Hive per esportare, importare o fare riferimento ai dati DynamoDB può portare a una perdita di precisione o a un errore nella query Hive.
Le esportazioni del tipo binary da DynamoDB ad Amazon Simple Storage Service (Amazon S3) o HDFS vengono memorizzate come una stringa con codifica Base64. Se importi dati da Amazon S3 o HDFS nel tipo binary di DynamoDB, assicurati che siano codificati come stringa Base64.
## Opzioni Hive
Puoi impostare le seguenti opzioni di Hive per gestire il trasferimento di dati da Amazon DynamoDB. Queste opzioni persistono nella sessione di Hive corrente. Se chiudi il prompt dei comandi Hive e lo riapri in un secondo momento nel cluster, le impostazioni ritorneranno ai valori predefiniti.
| Opzioni Hive | Description |
| --- | --- |
| dynamodb.throughput.read.percent | Imposta la percentuale di operazioni di lettura per mantenere il tasso di velocità effettiva assegnato di DynamoDB nell'intervallo allocato per la tabella. Il valore è compreso tra `0.1` e `1.5` inclusi. Il valore di 0,5 è la velocità di lettura di default, il che significa che Hive cercherà di consumare metà delle risorse di throughput assegnate alla lettura nella tabella. L'aumento del valore sopra a 0,5 aumenta la velocità della richiesta di lettura, che invece scende riducendo il valore sotto a 0,5. La velocità di lettura è approssimativa. La velocità di lettura reale varia in base a fattori come la presenza di una distribuzione uniforme delle chiavi in DynamoDB. Se vedi che il throughput assegnato viene frequentemente superato dalle operazioni Hive, oppure se è il traffico live in lettura viene limitato eccessivamente, porta il valore sotto a `0.5`. Se disponi di capacità sufficiente e desideri operazioni Hive più veloci, imposta questo valore al di sopra di `0.5`. Se ritieni che non ci siano operazioni di input/output inutilizzate disponibili, puoi anche andare oltre impostando il valore fino a 1,5. |
| dynamodb.throughput.write.percent | Imposta la percentuale di operazioni di scrittura per mantenere il tasso di velocità effettiva assegnato di DynamoDB nell'intervallo allocato per la tabella. Il valore è compreso tra `0.1` e `1.5` inclusi. Il valore di 0,5 è la velocità di scrittura di default, il che significa che Hive cercherà di consumare metà delle risorse di throughput assegnate alla scrittura nella tabella. L'aumento del valore sopra a 0,5 aumenta la velocità delle richieste di scrittura, che invece scende riducendo il valore sotto a 0,5. La velocità di scrittura è approssimativa. La velocità di scrittura reale varia in base a fattori come la presenza di una distribuzione uniforme delle chiavi in DynamoDB. Se vedi che il throughput assegnato viene frequentemente superato dalle operazioni Hive, oppure se è il traffico live in scrittura viene limitato eccessivamente, porta il valore sotto a `0.5`. Se disponi di capacità sufficiente e desideri operazioni Hive più veloci, imposta questo valore al di sopra di `0.5`. Puoi anche andare oltre impostando il valore fino a 1,5, se ritieni che non ci siano operazioni di input/output inutilizzate disponibili o se stai eseguendo il caricamento iniziale dei dati sulla tabella e non c'è ancora traffico live. |
| dynamodb.endpoint | Specifica l'endpoint per il servizio DynamoDB. Per ulteriori informazioni sugli endpoint DynamoDB disponibili, consulta la pagina relativa a [Regioni ed endpoint](https://docs.aws.amazon.com/general/latest/gr/rande.html#ddb_region). |
| dynamodb.max.map.tasks | Specifica il numero massimo di attività di mappatura durante la lettura di dati da DynamoDB. Questo valore deve essere maggiore o uguale a 1. |
| dynamodb.retry.duration | Specifica il numero di minuti da utilizzare come durata di timeout per rieseguire i comandi Hive. Questo valore deve essere un numero intero uguale o maggiore di 0. La durata di timeout predefinita è di due minuti. |
Queste opzioni vengono impostate utilizzando il comando `SET` come mostrato nel seguente esempio.
```
SET dynamodb.throughput.read.percent=1.0;
INSERT OVERWRITE TABLE s3_export SELECT *
FROM hiveTableName;
```
# Esempi di comandi Hive per l'esportazione, l'importazione e l'esecuzione di query sui dati in DynamoDB
Nei seguenti esempi vengono utilizzati comandi Hive per eseguire operazioni quali l'esportazione di dati su Amazon S3 o HDFS, l'importazione di dati in DynamoDB, l'unione di tabelle, l'esecuzione di query su tabelle e molto altro.
Le operazioni su una tabella Hive fanno riferimento ai dati archiviati in DynamoDB. I comandi di Hive sono soggetti alle impostazioni di velocità effettiva assegnate alla tabella DynamoDB e i dati recuperati includono i dati scritti nella tabella DynamoDB al momento dell'elaborazione della richiesta di operazione Hive da parte di DynamoDB. Se il processo di recupero dei dati richiede molto tempo, rispetto a quando è iniziato il comando Hive alcuni dei dati restituiti potrebbero essere stati aggiornati in DynamoDB.
I comandi Hive `DROP TABLE` e `CREATE TABLE` agiscono solo sulle tabelle locali in Hive e non creano né eliminano tabelle in DynamoDB. Se la query Hive fa riferimento a una tabella in DynamoDB, quest'ultima deve già esistere prima di eseguire la query. Per ulteriori informazioni sulla creazione e l'eliminazione di tabelle in DynamoDB, consulta [Lavorare con le tabelle in DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.html) nella *Guida per gli sviluppatori di Amazon DynamoDB*.
**Nota**
Quando mappi una tabella Hive su una posizione in Amazon S3, non mapparla al percorso principale del bucket, s3://amzn-s3-demo-bucket, poiché ciò potrebbe causare errori quando Hive scrive i dati su Amazon S3. Mappate invece la tabella su un sottopercorso del bucket, s3://amzn-s3-demo-bucket/mypath.
## Esportazione di dati da DynamoDB
È possibile utilizzare Hive per esportare i dati da DynamoDB.
**Esportazione di una tabella DynamoDB in un bucket Amazon S3**
+ Crea una tabella Hive che faccia riferimento ai dati archiviati in DynamoDB. A questo punto è possibile chiamare il comando INSERT OVERWRITE per scrivere i dati su una directory esterna. Nell'esempio seguente, *s3://amzn-s3-demo-bucket/path/subpath/* è un percorso valido in Amazon S3. Modifica le colonne e i tipi di dati nel comando CREATE (CREA) perché corrispondano ai valori in DynamoDB. È possibile utilizzare questo metodo per creare un archivio dei dati DynamoDB in Amazon S3.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. INSERT OVERWRITE DIRECTORY 's3://amzn-s3-demo-bucket/path/subpath/' SELECT *
7. FROM hiveTableName;
```
**Esportazione di una tabella DynamoDB in un bucket Amazon S3 utilizzando la formattazione**
+ Crea una tabella esterna che faccia riferimento a un percorso in Amazon S3. Questo passaggio è mostrato di seguito come s3\$1export. Durante la chiamata CREATE, specificare la formattazione della riga per la tabella. Quindi, quando si utilizza INSERT OVERWRITE per esportare i dati da DynamoDB in s3\$1export, i dati vengono scritti nel formato specificato. In questo esempio, i dati vengono scritti come valori separati da virgola (CSV).
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. CREATE EXTERNAL TABLE s3_export(a_col string, b_col bigint, c_col array)
7. ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
8. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
9.
10. INSERT OVERWRITE TABLE s3_export SELECT *
11. FROM hiveTableName;
```
**Esportazione di una tabella DynamoDB in un bucket Amazon S3 senza specificare una mappatura di colonna**
+ Crea una tabella Hive che faccia riferimento ai dati archiviati in DynamoDB. L'operazione è analoga a quella dell'esempio precedente, salvo che non si specifica una mappatura di colonna. La tabella deve avere esattamente una colonna di tipo `map`. Se si crea una tabella `EXTERNAL` in Amazon S3 è possibile chiamare il comando `INSERT OVERWRITE` per scrivere i dati provenienti da DynamoDB su Amazon S3. È possibile utilizzare questo metodo per creare un archivio dei dati DynamoDB in Amazon S3. Poiché non è presente alcuna mappatura di colonna, non è possibile eseguire query in tabelle esportate in questo modo. L'esportazione dei dati senza specificare una mappatura di colonna è disponibile in Hive versione 0.8.1.5 o successive, supportate sull'AMI Amazon EMR versione 2.2.*x* e successive.
```
1. CREATE EXTERNAL TABLE hiveTableName (item map)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1");
4.
5. CREATE EXTERNAL TABLE s3TableName (item map)
6. ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
7. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
8.
9. INSERT OVERWRITE TABLE s3TableName SELECT *
10. FROM hiveTableName;
```
**Esportazione di una tabella DynamoDB in un bucket Amazon S3 utilizzando la compressione dei dati**
+ Hive fornisce diversi codec di compressione che è possibile impostare durante la sessione Hive. In questo modo, i dati esportati vengono compressi nel formato specificato. L'esempio seguente comprime i file esportati utilizzando l'algoritmo Lempel-Ziv-Oberhumer (LZO).
```
1. SET hive.exec.compress.output=true;
2. SET io.seqfile.compression.type=BLOCK;
3. SET mapred.output.compression.codec = com.hadoop.compression.lzo.LzopCodec;
4.
5. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
6. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
7. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
8. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
9.
10. CREATE EXTERNAL TABLE lzo_compression_table (line STRING)
11. ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
12. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
13.
14. INSERT OVERWRITE TABLE lzo_compression_table SELECT *
15. FROM hiveTableName;
```
I codec di compressione disponibili sono:
+ org.apache.hadoop.io.compress. GzipCodec
+ org.apache.hadoop.io.compress. DefaultCodec
+ com.hadoop.compression.lzo. LzoCodec
+ com.hadoop.compression.lzo. LzopCodec
+ org.apache.hadoop.io.compress. BZip2Codec
+ org.apache.hadoop.io.compress. SnappyCodec
**Esportazione di una tabella DynamoDB in HDFS**
+ Usa il seguente comando Hive, dove *hdfs:///directoryName* è un percorso HDFS valido ed *hiveTableName* è una tabella in Hive che fa riferimento a DynamoDB. Questa operazione di esportazione è più veloce dell'esportazione di una tabella DynamoDB su Amazon S3, perché Hive 0.7.1.1 usa HDFS come fase intermediaria per l'esportazione di dati su Amazon S3. L'esempio seguente mostra anche come impostare il valore `dynamodb.throughput.read.percent` su 1,0 per aumentare la richiesta di lettura.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. SET dynamodb.throughput.read.percent=1.0;
7.
8. INSERT OVERWRITE DIRECTORY 'hdfs:///directoryName' SELECT * FROM hiveTableName;
```
È anche possibile esportare i dati su HDFS utilizzando la formattazione e la compressione come illustrato sopra per l'esportazione ad Amazon S3. Per farlo, è sufficiente sostituire la directory Amazon S3 negli esempi sopra con una directory HDFS.
**Per leggere dati di caratteri UTF-8 non stampabili in Hive**
+ È possibile leggere e scrivere dati di caratteri UTF-8 non stampabili con Hive utilizzando la clausola `STORED AS SEQUENCEFILE` nella creazione di una tabella. A SequenceFile è il formato di file binario Hadoop; è necessario utilizzare Hadoop per leggere questo file. Nell'esempio seguente viene illustrato come esportare dati da DynamoDB verso Amazon S3. Puoi utilizzare questa funzionalità per gestire i caratteri in codifica UTF-8 non stampabili.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. CREATE EXTERNAL TABLE s3_export(a_col string, b_col bigint, c_col array)
7. STORED AS SEQUENCEFILE
8. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
9.
10. INSERT OVERWRITE TABLE s3_export SELECT *
11. FROM hiveTableName;
```
## Importazione di dati in DynamoDB
Quando scrivi dati in DynamoDB utilizzando Hive, devi assicurarti che il numero delle unità di capacità di scrittura sia maggiore del numero di mappatori del cluster. Ad esempio, i cluster in esecuzione su istanze EC2 m1.xlarge producono 8 mappatori per istanza. Nel caso di un cluster che dispone di 10 istanze, ciò significherebbe un totale di 80 mappatori. Se il numero di unità di capacità di scrittura non è superiore al numero di mappatori nel cluster, l'operazione di scrittura Hive potrebbe consumare tutta la velocità effettiva di scrittura o tentare di consumare più velocità effettiva di quanto ne sia stato assegnato. Per ulteriori informazioni sul numero di mappatori creato da ogni tipo di istanza EC2, consulta [Configura Hadoop](emr-hadoop-config.md).
Il numero di mappatori in Hadoop è controllato dalle divisioni in entrata. Se le divisioni sono troppo poche, il comando di scrittura potrebbe non essere in grado di consumare tutto il throughput di scrittura disponibile.
Se un elemento con la stessa chiave è presente nella tabella DynamoDB di destinazione, viene sovrascritto. Se un elemento con la stessa chiave non è presente nella tabella DynamoDB di destinazione, viene inserito.
**Importazione di una tabella da Amazon S3 a DynamoDB**
+ È possibile utilizzare Amazon EMR e Hive per scrivere i dati da Amazon S3 a DynamoDB.
```
CREATE EXTERNAL TABLE s3_import(a_col string, b_col bigint, c_col array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
INSERT OVERWRITE TABLE hiveTableName SELECT * FROM s3_import;
```
**Importazione una tabella da un bucket Amazon S3 a DynamoDB senza specificare una mappatura di colonna**
+ Crea una tabella `EXTERNAL` che faccia riferimento a dati archiviati in Amazon S3 ed esportati in precedenza da DynamoDB. Prima dell'importazione, assicurati che la tabella sia presente in DynamoDB e che abbia lo stesso schema di chiave della tabella DynamoDB esportata in precedenza. La tabella deve inoltre avere esattamente una colonna di tipo `map`. Se si crea una tabella Hive collegata a DynamoDB è possibile chiamare il comando `INSERT OVERWRITE` per scrivere i dati provenienti da Amazon S3 su DynamoDB. Poiché non è presente alcuna mappatura di colonna, non è possibile eseguire query in tabelle importate in questo modo. L'importazione dei dati senza specificare una mappatura di colonna è disponibile in Hive versione 0.8.1.5 o successive, supportate sull'AMI Amazon EMR 2.2.3 e successive.
```
CREATE EXTERNAL TABLE s3TableName (item map)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
CREATE EXTERNAL TABLE hiveTableName (item map)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1");
INSERT OVERWRITE TABLE hiveTableName SELECT *
FROM s3TableName;
```
**Importazione di una tabella da HDFS a DynamoDB**
+ È possibile utilizzare Amazon EMR e Hive per scrivere i dati da HDFS a DynamoDB.
```
CREATE EXTERNAL TABLE hdfs_import(a_col string, b_col bigint, c_col array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 'hdfs:///directoryName';
CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
INSERT OVERWRITE TABLE hiveTableName SELECT * FROM hdfs_import;
```
## Esecuzioni di query sui dati in DynamoDB
Negli esempi seguenti vengono illustrati i diversi modi in cui puoi utilizzare Amazon EMR per eseguire query sui dati archiviati in DynamoDB.
**Per trovare il valore maggiore per una colonna mappata (`max`)**
+ Utilizzare comandi Hive come quelli indicati di seguito. Nel primo comando l'istruzione CREATE (CREA) crea una tabella Hive che faccia riferimento ai dati archiviati in DynamoDB. L'istruzione SELECT usa quindi la tabella per eseguire una query sui dati archiviati in DynamoDB. L'esempio seguente trova l'ordine più grande effettuato da un determinato cliente.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
SELECT max(total_cost) from hive_purchases where customerId = 717;
```
**Per aggregare i dati utilizzando la clausola `GROUP BY`**
+ Puoi utilizzare la clausola `GROUP BY` per raccogliere dati all'interno di più record. Questa clausola viene spesso utilizzata con una funzione di aggregazione come sum, count, min o max. L'esempio seguente restituisce un elenco degli ordini più grandi di clienti che hanno effettuato più di tre ordini.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
SELECT customerId, max(total_cost) from hive_purchases GROUP BY customerId HAVING count(*) > 3;
```
**Unione di due tabelle DynamoDB**
+ L'esempio seguente mappa due tabelle Hive ai dati archiviati in DynamoDB. quindi chiama un'operazione di unione tra le due tabelle. L'unione viene calcolata nel cluster e in seguito viene restituita. L'unione non si svolge in DynamoDB. Questo esempio restituisce un elenco di clienti e dei loro acquisti per i clienti che hanno effettuato più di due ordini.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
CREATE EXTERNAL TABLE hive_customers(customerId bigint, customerName string, customerAddress array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Customers",
"dynamodb.column.mapping" = "customerId:CustomerId,customerName:Name,customerAddress:Address");
Select c.customerId, c.customerName, count(*) as count from hive_customers c
JOIN hive_purchases p ON c.customerId=p.customerId
GROUP BY c.customerId, c.customerName HAVING count > 2;
```
**Per unire due tabelle da origini diverse**
+ In questo esempio, Customer\$1S3 è una tabella Hive che carica un file CSV archiviato in Amazon S3, mentre hive\$1purchases è una tabella che fa riferimento ai dati in DynamoDB. L'esempio seguente unisce i dati dei clienti memorizzati come file CSV in Amazon S3 con i dati degli ordini archiviati in DynamoDB per restituire un set di dati che rappresenta gli ordini effettuati dai clienti il cui nome contiene la parola "Miller".
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
CREATE EXTERNAL TABLE Customer_S3(customerId bigint, customerName string, customerAddress array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
Select c.customerId, c.customerName, c.customerAddress from
Customer_S3 c
JOIN hive_purchases p
ON c.customerid=p.customerid
where c.customerName like '%Miller%';
```
**Nota**
In esempi precedenti, le istruzioni CREATE TABLE erano incluse in ciascun esempio per chiarezza e completezza. Se vengono eseguite più operazioni di query o di esportazione su una determinata tabella Hive, è necessario creare la tabella solo una sola volta, all'inizio della sessione Hive.
# Ottimizzazione delle prestazioni per le operazioni Amazon EMR in DynamoDB
Le operazioni Amazon EMR su una tabella DynamoDB vengono conteggiate come operazioni di lettura e sono soggette alle impostazioni di velocità effettiva assegnata della tabella. Amazon EMR implementa una propria logica per provare a bilanciare il carico sulla tabella DynamoDB e ridurre al minimo la possibilità di superare la velocità effettiva assegnata. Al termine di ogni query Hive, Amazon EMR restituisce informazioni sul cluster utilizzato per elaborare la query, incluso il numero di volte in cui la velocità effettiva assegnata è stata superata. È possibile utilizzare queste informazioni, oltre alle CloudWatch metriche sul throughput di DynamoDB, per gestire meglio il carico sulla tabella DynamoDB nelle richieste successive.
Nelle operazioni con le tabelle DynamoDB, le prestazioni delle query di Hive sono influenzate dai fattori indicati di seguito.
## Unità di capacità di lettura assegnate
Quando esegui query Hive su una tabella DynamoDB, devi accertarti di aver assegnato una quantità sufficiente di unità di capacità di lettura.
Ad esempio, supponiamo che tu disponga di 100 unità di capacità di lettura assegnate per la tabella DynamoDB. Questo ti consentirà di eseguire 100 operazioni di lettura, o 409.600 byte, al secondo. Se la tabella contiene 20 GB di dati (21.474.836.480 byte) e la tua query Hive esegue una scansione completa della tabella, puoi stimare la durata dell'esecuzione della query:
* 21.474.836.480 / 409.600 = 52.429 secondi = 14,56 ore *
Il solo modo per ridurre il tempo necessario sarebbe modificare le unità di capacità di lettura nella tabella DynamoDB di origine. L'aggiunta di più nodi al cluster Amazon EMR non è utile.
Nell'output Hive, la percentuale di completamento viene aggiornata quando terminano uno o più processi del mappatore. Per una tabella DynamoDB di grandi dimensioni con una bassa capacità di lettura assegnata, l'output della percentuale di completamento potrebbe non essere aggiornato per molto tempo; in questo caso, il processo sembrerà essere completo allo 0% per diverse ore. Per uno stato più dettagliato dell'avanzamento del processo, accedi alla console di Amazon EMR, dalla quale potrai visualizzare lo stato delle singole attività del mappatore e statistiche per le letture di dati.
Puoi anche accedere all'interfaccia Hadoop sul nodo master e vedere le statistiche di Hadoop. Ti sarà mostrato lo stato della singola attività del mappatore e alcune statistiche di lettura dei dati. Per ulteriori informazioni, consulta la sezione relativa alle [Interfacce Web ospitate sul nodo master](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html) nella *Guida alla gestione di Amazon EMR*.
## Impostazione della percentuale di lettura
Per impostazione predefinita, Amazon EMR gestisce il carico di richieste sulla tabella DynamoDB, in base alla velocità effettiva corrente assegnata. Tuttavia, se Amazon EMR restituisce informazioni sul processo in cui è incluso un elevato numero di risposte di superamento della velocità effettiva assegnata, puoi modificare la velocità di lettura di default utilizzando il parametro `dynamodb.throughput.read.percent` durante la configurazione della tabella Hive. Per ulteriori informazioni su come impostare il parametro della percentuale di lettura, consulta [Opzioni Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Impostazione della percentuale di scrittura
Per impostazione predefinita, Amazon EMR gestisce il carico di richieste sulla tabella DynamoDB, in base alla velocità effettiva corrente assegnata. Tuttavia, se Amazon EMR restituisce informazioni sul processo in cui è incluso un elevato numero di risposte di superamento della velocità effettiva assegnata, puoi modificare la velocità di scrittura di default utilizzando il parametro `dynamodb.throughput.write.percent` durante la configurazione della tabella Hive. Per ulteriori informazioni su come impostare il parametro della percentuale di scrittura, consulta [Opzioni Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Impostazione della durata dei nuovi tentativi
Per impostazione predefinita, Amazon EMR esegue nuovamente una query Hive se questa non ha restituito un risultato entro due minuti, l'intervallo predefinito per i nuovi tentativi. Puoi modificare questo intervallo impostando il parametro `dynamodb.retry.duration` quando esegui una query Hive. Per ulteriori informazioni su come impostare il parametro della percentuale di scrittura, consulta [Opzioni Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Numero di attività di mappatura
I daemon di mappatura che Hadoop avvia per elaborare le richieste di esportazione ed esecuzione di query sui dati archiviati in DynamoDB hanno una velocità di lettura massima di 1 MiB al secondo per limitare la capacità di lettura utilizzata. Se disponi di ulteriore velocità effettiva assegnata disponibile su DynamoDB, puoi migliorare le prestazioni delle operazioni di query ed esportazione di Hive aumentando il numero di daemon del mappatore. A questo scopo, puoi aumentare il numero di istanze EC2 nel cluster *o* il numero di daemon del mappatore in esecuzione su ciascuna istanza EC2.
Puoi aumentare il numero di istanze EC2 in un cluster arrestando il cluster corrente e riavviandolo con un numero maggiore di istanze EC2. Puoi specificare il numero di istanze EC2 nella finestra di dialogo **Configure EC2 Instances (Configura istanze EC2)** se avvii il cluster dalla console di Amazon EMR o con l'opzione `‑‑num-instances` se lo avvii dalla CLI.
Il numero di attività di mappatura eseguite su un'istanza dipende dal tipo di istanza EC2. Per ulteriori informazioni sui tipi di istanze EC2 supportate e sul numero di mappatori fornito da ognuno, consulta [Configurazione attività](emr-hadoop-task-config.md), dove troverai una sezione relativa alla configurazione delle attività per ognuna delle configurazioni supportate.
In alternativa, puoi aumentare il numero dei daemon di mappatura modificando il parametro di configurazione `mapreduce.tasktracker.map.tasks.maximum` di Hadoop con un valore superiore. Questo metodo presenta il vantaggio di offrirti un numero maggiore di mappatori senza aumentare né il numero, né le dimensioni delle istanze EC2, con conseguente risparmio di denaro. Lo svantaggio è che un'impostazione troppo elevata in questo valore può provocare l'esaurimento della memoria nelle istanze EC2 del tuo cluster. Per impostare `mapreduce.tasktracker.map.tasks.maximum`, avvia il cluster e specifica un valore per `mapreduce.tasktracker.map.tasks.maximum` come proprietà della classificazione di configurazione mapred-site. Questo viene mostrato nell’esempio seguente. Per ulteriori informazioni, consulta [Configurazione delle applicazioni](emr-configure-apps.md).
```
{
"configurations": [
{
"classification": "mapred-site",
"properties": {
"mapred.tasktracker.map.tasks.maximum": "10"
}
}
]
}
```
## Richieste di dati in parallelo
Molteplici richieste di dati, sia da parte di più utenti sia da più applicazioni verso un'unica tabella, possono far esaurire il throughput di lettura assegnato e rallentare le prestazioni.
## Durata dei processi
La consistenza dei dati in DynamoDB dipende dall'ordine delle operazioni di lettura e scrittura di ciascun nodo. Quando una query Hive è in avanzamento, un'altra applicazione potrebbe caricare nuovi dati nella tabella DynamoDB oppure modificare o eliminare dati esistenti. In questo caso, i risultati della query Hive potrebbe non riflettere le modifiche effettuate ai dati durante l'esecuzione della query.
## Evitare di superare la velocità effettiva
Quando esegui query Hive su DynamoDB, fai attenzione a non superare la velocità effettiva assegnata, perché in questo modo si esaurisce la capacità necessaria per le chiamate dell'applicazione a `DynamoDB::Get`. Per evitare che ciò accada, dovresti monitorare regolarmente il volume di lettura e la limitazione delle chiamate alle applicazioni controllando i log e `DynamoDB::Get` monitorando le metriche su Amazon. CloudWatch
## Ora delle richieste
Le prestazioni possono essere migliorate pianificando query Hive che accedono a una tabella DynamoDB quando la richiesta nella tabella DynamoDB è minore. Ad esempio, se la maggior parte degli utenti dell'applicazione vive a San Francisco, è possibile scegliere di esportare i dati ogni giorno alle ore 4:00 (PST), quando la maggior parte degli utenti è inattiva e non aggiorna i registri del database DynamoDB.
## Tabelle basate sul tempo
Se i dati vengono organizzati in una serie di tabelle DynamoDB basate sul tempo, ad esempio una tabella al giorno, puoi esportare i dati quando la tabella non è più attiva. Puoi utilizzare questa tecnica per eseguire il backup dei dati su Amazon S3 in modo regolare.
## Dati archiviati
Se prevedi di eseguire molte query Hive sui dati archiviati in DynamoDB e la tua applicazione è in grado di tollerare i dati archiviati, potrebbe essere una buona idea esportare i dati su HDFS o Amazon S3 ed eseguire le query Hive su una copia dei dati, anziché su DynamoDB. In questo modo, le operazioni di lettura e il throughput assegnato vengono conservati.
# Kinesis
I cluster Amazon EMR possono leggere ed elaborare direttamente gli stream Amazon Kinesis, utilizzando strumenti familiari nell'ecosistema Hadoop come Hive, Pig, l'API Hadoop Streaming e Cascading. MapReduce È anche possibile unire i dati in tempo reale provenienti da Amazon Kinesis con dati esistenti su Amazon S3, Amazon DynamoDB e HDFS in un cluster in esecuzione. I dati possono essere caricati direttamente da Amazon EMR su Amazon S3 o DynamoDB per le attività di post-elaborazione. Per ulteriori informazioni sulle caratteristiche salienti del servizio Amazon Kinesis e sui prezzi, consulta la pagina di [Amazon Kinesis](https://aws.amazon.com//kinesis).
## Cosa posso fare con l'integrazione di Amazon EMR e Amazon Kinesis?
L'integrazione tra Amazon EMR e Amazon Kinesis semplifica notevolmente alcuni scenari, quali:
+ **Analisi dei log di streaming**: puoi analizzare i log Web di streaming per generare ogni pochi minuti un elenco dei 10 errori più diffusi in base alla regione, al browser e al dominio di accesso.
+ **Coinvolgimento dei clienti**: puoi scrivere query che uniscano i dati clickstream da Amazon Kinesis con le informazioni sulle campagne pubblicitarie archiviate in una tabella DynamoDB al fine di identificare le categorie di annunci più efficaci tra quelli visualizzati su specifici siti Web.
+ **Query interattive ad-hoc**: puoi caricare periodicamente dati dai flussi Amazon Kinesis in HDFS e renderli disponibili come tabella locale Impala per query rapide, interattive e analitiche.
## Analisi con checkpoint dei flussi Amazon Kinesis
Gli utenti possono eseguire analisi periodiche in batch dei flussi Amazon Kinesis nelle cosiddette *iterazioni*. Poiché i registri dei dati di flusso Amazon Kinesis vengono recuperati utilizzando un numero di sequenza, i limiti dell'iterazione sono definiti da numeri di sequenza iniziali e finali che Amazon EMR archivia in una tabella DynamoDB. Ad esempio, al termine di `iteration0`, il numero in sequenza finale viene archiviato in DynamoDB in modo che, all'inizio del processo `iteration1`, sia possibile recuperare i dati successivi dal flusso. Questa mappatura delle iterazioni nei dati di flusso si definisce *checkpoint*. Per ulteriori informazioni, consulta [Kinesis Connector](https://aws.amazon.com/elasticmapreduce/faqs/#kinesis-connector).
Se per un'iterazione sono stati definiti checkpoint e il processo non ha potuto elaborare un'iterazione, Amazon EMR tenta di rielaborare il record in quella iterazione.
I checkpoint consentono di:
+ Avviare l'elaborazione dei dati dopo un numero di sequenza elaborato da una precedente query eseguita sullo stesso flusso e nome logico
+ Rielaborazione dello stesso batch di dati Kinesis elaborato da una precedente query
Per abilitare i checkpoint, imposta il parametro `kinesis.checkpoint.enabled` su `true` e configura i seguenti parametri:
| Impostazione di configurazione | Description |
| --- | --- |
| kinesis.checkpoint.metastore.table.name | Nome della tabella DynamoDB in cui saranno archiviate le informazioni dei checkpoint |
| kinesis.checkpoint.metastore.hash.key.name | Nome chiave hash per la tabella DynamoDB |
| kinesis.checkpoint.metastore.hash.range.name | Nome chiave range per la tabella DynamoDB |
| kinesis.checkpoint.logical.name | Un nome logico per l'elaborazione corrente |
| kinesis.checkpoint.iteration.no | Numero di iterazioni per elaborazione associata al nome logico |
| kinesis.rerun.iteration.without.wait | Valore booleano che indica se un'iterazione non riuscita può essere eseguita nuovamente senza timeout; l'impostazione predefinita è false |
### Suggerimenti relativi a Provisioned IOPS per le tabelle Amazon DynamoDB
Il connettore Amazon EMR per Amazon Kinesis utilizza il database DynamoDB come database di supporto per il checkpoint dei metadati. Devi creare una tabella in DynamoDB prima di consumare i dati in un flusso Amazon Kinesis con un cluster Amazon EMR in intervalli sui cui è stato eseguito il checkpoint. La tabella deve trovarsi nella stessa regione del cluster Amazon EMR. Di seguito sono riportate le raccomandazioni generali per il numero di IOPS da assegnare alle tabelle DynamoDB, con `j` che sarà il numero massimo di processi Hadoop (con diverse combinazioni nome logico\$1numero di iterazione) che possono essere eseguiti simultaneamente e `s` il numero massimo di shard elaborati da qualsiasi processo:
Per **Read Capacity Units (Unità di capacità di lettura)**: `j`\$1`s`/`5`
Per **Write Capacity Units (Unità di capacità di scrittura)**: `j`\$1`s`
## Considerazioni sulle prestazioni
La velocità effettiva degli shard Amazon Kinesis è direttamente proporzionale alle dimensioni dell'istanza dei nodi nei cluster Amazon EMR e alle dimensioni dei record nel flusso. Consigliamo di utilizzare l'istanza m5.xlarge o istanze più grandi sui nodi master e principali.
## Pianificazione dell'analisi Amazon Kinesis con Amazon EMR
Quando analizzi i dati su un flusso Amazon Kinesis attivo, dati i limiti imposti dal timeout e da una durata massima per ogni iterazione, è importante eseguire l'analisi di frequente per raccogliere dettagli periodici dal flusso. Esistono diversi modi per eseguire tali script e query a intervalli regolari; consigliamo l'uso di AWS Data Pipeline per attività ricorrenti come queste. Per ulteriori [AWS Data Pipeline HiveActivity](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-hiveactivity.html)informazioni *AWS Data Pipeline , consulta [AWS Data Pipeline PigActivity](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-pigactivity.html)*e consulta la Developer Guide.
# Migrazione del connettore Spark Kinesis all'SDK 2.x per Amazon EMR 7.0
L' AWS SDK offre un ricco set di librerie per interagire con APIs i servizi di AWS cloud computing, come la gestione delle credenziali, la connessione ai servizi S3 e Kinesis. Il connettore Spark Kinesis viene utilizzato per consumare dati dal flusso di dati Kinesis e i dati ricevuti vengono trasformati ed elaborati nel motore di esecuzione di Spark. Attualmente questo connettore è basato su 1.x di AWS SDK e (KCL). Kinesis-client-library
Come parte della migrazione all' AWS SDK 2.x, anche il connettore Spark Kinesis viene aggiornato di conseguenza per funzionare con l'SDK 2.x. Nella versione 7.0 di Amazon EMR, Spark contiene l'aggiornamento SDK 2.x che non è ancora disponibile nella versione community di Apache Spark. Se utilizzi il connettore Spark Kinesis da una versione precedente alla 7.0, devi effettuare la migrazione dei codici dell'applicazione per eseguirli su SDK 2.x prima di poterla effettuare su Amazon EMR 7.0.
## Guide alla migrazione
Questa sezione descrive i passaggi per eseguire la migrazione di un'applicazione al connettore Spark Kinesis aggiornato. Include guide per la migrazione alla Kinesis Client Library (KCL) 2.x AWS , fornitori di credenziali AWS e client di servizi in SDK 2.x. AWS A titolo di riferimento, include anche un [WordCount](https://github.com/apache/spark/blob/v3.5.0/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala)programma di esempio che utilizza il connettore Kinesis.
**Topics**
+ [Migrazione di KLC da 1.x a 2.x](#migrating-spark-kinesis-KCL-from-1.x-to-2.x)
+ [Migrazione dei fornitori di AWS credenziali da SDK 1.x a 2.x AWS](#migrating-spark-kinesis-creds-from-1.x-to-2.x)
+ [Migrazione dei client di AWS servizio da AWS SDK 1.x a 2.x](#migrating-spark-kinesis-service-from-1.x-to-2.x)
+ [Esempi di codice per applicazioni di streaming](#migrating-spark-kinesis-streaming-examples)
+ [Considerazioni sull'utilizzo del connettore Spark Kinesis aggiornato](#migrating-spark-kinesis-considerations)
### Migrazione di KLC da 1.x a 2.x
+ **Parametri, livello e dimensioni in `KinesisInputDStream`**
Quando crei un'istanza `KinesisInputDStream`, puoi controllare il livello e le dimensioni dei parametri per il flusso. L'esempio seguente mostra come personalizzare questi parametri con KCL 1.x:
```
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
val kinesisStream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet)
.build()
```
In KCL 2.x, queste impostazioni di configurazione hanno nomi di pacchetto diversi. Per eseguire la migrazione a 2.x:
1. Modifica le istruzioni di importazione per `com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration` e `com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel` rispettivamente in `software.amazon.kinesis.metrics.MetricsLevel` e `software.amazon.kinesis.metrics.MetricsUtil`.
```
// import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
import software.amazon.kinesis.metrics.MetricsLevel
// import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
import software.amazon.kinesis.metrics.MetricsUtil
```
1. Sostituisci la riga `metricsEnabledDimensionsKinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet` con `metricsEnabledDimensionsSet(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME)`
Di seguito è riportata una versione aggiornata di `KinesisInputDStream` con livello e dimensioni dei parametri personalizzati:
```
import software.amazon.kinesis.metrics.MetricsLevel
import software.amazon.kinesis.metrics.MetricsUtil
val kinesisStream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME))
.build()
```
+ Funzione del gestore di messaggi in `KinesisInputDStream`
Quando crei un'istanza `KinesisInputDStream`, puoi anche fornire una "funzione del gestore di messaggi" che accetta un Kinesis Record e restituisce un oggetto generico T, nel caso in cui desideri utilizzare altri dati inclusi in un Record come la chiave di partizione.
In KCL 1.x, la firma della funzione del gestore di messaggi è `Record => T`, dove Record è `com.amazonaws.services.kinesis.model.Record`. In KCL 2.x, la firma del gestore viene modificata in:`KinesisClientRecord => T`, where is. KinesisClientRecord `software.amazon.kinesis.retrieval.KinesisClientRecord`
Di seguito è riportato un esempio di fornitura di un gestore di messaggi in KCL 1.x:
```
import com.amazonaws.services.kinesis.model.Record
def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
val stream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.buildWithMessageHandler(addFive)
```
Per eseguire la migrazione del gestore di messaggi:
1. Modifica l'istruzione di importazione per `com.amazonaws.services.kinesis.model.Record` in `software.amazon.kinesis.retrieval.KinesisClientRecord`.
```
// import com.amazonaws.services.kinesis.model.Record
import software.amazon.kinesis.retrieval.KinesisClientRecord
```
1. Aggiorna la firma del metodo del gestore di messaggi.
```
//def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
def addFive = (r: KinesisClientRecord) => JavaUtils.bytesToString(r.data()).toInt + 5
```
Di seguito è riportato un esempio aggiornato di fornitura di un gestore di messaggi in KCL 2.x:
```
import software.amazon.kinesis.retrieval.KinesisClientRecord
def addFive = (r: KinesisClientRecord) => JavaUtils.bytesToString(r.data()).toInt + 5
val stream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.buildWithMessageHandler(addFive)
```
Per ulteriori informazioni sulla migrazione da KCL 1.x a 2.x, consulta [Migrazione dei consumatori da KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html).
### Migrazione dei fornitori di AWS credenziali da SDK 1.x a 2.x AWS
I fornitori di credenziali vengono utilizzati per ottenere credenziali per le interazioni con. AWS AWS Esistono diverse modifiche all'interfaccia e alla classe relative ai fornitori di credenziali in SDK 2.x, che sono disponibili [qui](https://github.com/aws/aws-sdk-java-v2/blob/master/docs/LaunchChangelog.md#122-client-credentials). Il connettore Spark Kinesis ha definito un'interfaccia `org.apache.spark.streaming.kinesis.SparkAWSCredentials` () e classi di implementazione che restituiscono la versione AWS 1.x dei provider di credenziali. Questi fornitori di credenziali sono necessari per inizializzare i client Kinesis. Ad esempio, se utilizzi il metodo `SparkAWSCredentials.provider` nelle applicazioni, dovrai aggiornare i codici per utilizzare la versione 2.x dei provider di credenziali. AWS
Di seguito è riportato un esempio di utilizzo dei provider di credenziali in AWS SDK 1.x:
```
import org.apache.spark.streaming.kinesis.SparkAWSCredentials
import com.amazonaws.auth.AWSCredentialsProvider
val basicSparkCredentials = SparkAWSCredentials.builder
.basicCredentials("accessKey", "secretKey")
.build()
val credentialProvider = basicSparkCredentials.provider
assert(credentialProvider.isInstanceOf[AWSCredentialsProvider], "Type should be AWSCredentialsProvider")
```
**Per eseguire la migrazione a SDK 2.x:**
1. Modifica l'istruzione di importazione per `com.amazonaws.auth.AWSCredentialsProvider` in `software.amazon.awssdk.auth.credentials.AwsCredentialsProvider`
```
//import com.amazonaws.auth.AWSCredentialsProvider
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
```
1. Aggiorna i codici rimanenti che utilizzano questa classe.
```
import org.apache.spark.streaming.kinesis.SparkAWSCredentials
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
val basicSparkCredentials = SparkAWSCredentials.builder
.basicCredentials("accessKey", "secretKey")
.build()
val credentialProvider = basicSparkCredentials.provider
assert (credentialProvider.isInstanceOf[AwsCredentialsProvider], "Type should be AwsCredentialsProvider")
```
### Migrazione dei client di AWS servizio da AWS SDK 1.x a 2.x
AWS i client di servizio hanno nomi di pacchetto diversi in 2.x (cioè`software.amazon.awssdk`), mentre l'SDK 1.x lo utilizza. `com.amazonaws` Per ulteriori informazioni sulle modifiche del client, consulta [questa pagina](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html). Se utilizzi questi client del servizio nei codici, devi eseguire la migrazione dei client di conseguenza.
Di seguito è riportato un esempio di creazione di un client in SDK 1.x:
```
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
import com.amazonaws.services.dynamodbv2.document.DynamoDB
AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
```
**Per eseguire la migrazione a 2.x:**
1. Modifica le istruzioni di importazione per i client del servizio. Prendiamo ad esempio i client DynamoDB. Devi cambiare `com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient` o `com.amazonaws.services.dynamodbv2.document.DynamoDB` in `software.amazon.awssdk.services.dynamodb.DynamoDbClient`.
```
// import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
// import com.amazonaws.services.dynamodbv2.document.DynamoDB
import software.amazon.awssdk.services.dynamodb.DynamoDbClient
```
1. Aggiorna i codici che inizializzano i client
```
// AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
// AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
DynamoDbClient ddbClient = DynamoDbClient.create();
DynamoDbClient ddbClient = DynamoDbClient.builder().build();
```
Per ulteriori informazioni sulla migrazione dell' AWS SDK da 1.x a 2.x, consulta [Cosa c'è di diverso tra SDK for AWS Java](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html) 1.x e 2.x
### Esempi di codice per applicazioni di streaming
```
import java.net.URI
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.services.kinesis.KinesisClient
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest
import software.amazon.awssdk.regions.Region
import software.amazon.kinesis.metrics.{MetricsLevel, MetricsUtil}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.kinesis.KinesisInputDStream
object KinesisWordCountASLSDKV2 {
def main(args: Array[String]): Unit = {
val appName = "demo-app"
val streamName = "demo-kinesis-test"
val endpointUrl = "https://kinesis.us-west-2.amazonaws.com"
val regionName = "us-west-2"
// Determine the number of shards from the stream using the low-level Kinesis Client
// from the AWS Java SDK.
val credentialsProvider = DefaultCredentialsProvider.create
require(credentialsProvider.resolveCredentials() != null,
"No AWS credentials found. Please specify credentials using one of the methods specified " +
"in https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html")
val kinesisClient = KinesisClient.builder()
.credentialsProvider(credentialsProvider)
.region(Region.US_WEST_2)
.endpointOverride(URI.create(endpointUrl))
.httpClientBuilder(ApacheHttpClient.builder())
.build()
val describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build()
val numShards = kinesisClient.describeStream(describeStreamRequest)
.streamDescription
.shards
.size
// In this example, we are going to create 1 Kinesis Receiver/input DStream for each shard.
// This is not a necessity; if there are less receivers/DStreams than the number of shards,
// then the shards will be automatically distributed among the receivers and each receiver
// will receive data from multiple shards.
val numStreams = numShards
// Spark Streaming batch interval
val batchInterval = Milliseconds(2000)
// Kinesis checkpoint interval is the interval at which the DynamoDB is updated with information
// on sequence number of records that have been received. Same as batchInterval for this
// example.
val kinesisCheckpointInterval = batchInterval
// Setup the SparkConfig and StreamingContext
val sparkConfig = new SparkConf().setAppName("KinesisWordCountASLSDKV2")
val ssc = new StreamingContext(sparkConfig, batchInterval)
// Create the Kinesis DStreams
val kinesisStreams = (0 until numStreams).map { i =>
KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME))
.build()
}
// Union all the streams
val unionStreams = ssc.union(kinesisStreams)
// Convert each line of Array[Byte] to String, and split into words
val words = unionStreams.flatMap(byteArray => new String(byteArray).split(" "))
// Map each word to a (word, 1) tuple so we can reduce by key to count the words
val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
// Print the first 10 wordCounts
wordCounts.print()
// Start the streaming context and await termination
ssc.start()
ssc.awaitTermination()
}
}
```
### Considerazioni sull'utilizzo del connettore Spark Kinesis aggiornato
+ Se le tue applicazioni utilizzano la versione `Kinesis-producer-library` con la versione di JDK precedente alla 11, potresti imbatterti in eccezioni come `java.lang.NoClassDefFoundError: javax/xml/bind/DatatypeConverter`. Ciò accade perché EMR 7.0 viene fornito con JDK 17 per impostazione predefinita e i moduli J2EE sono stati rimossi dalle librerie standard a partire da Java 11\$1. Questo problema può essere risolto aggiungendo la seguente dipendenza nel file pom. Sostituisci la versione della libreria con una versione che pensi possa essere più adeguata.
```
javax.xml.bind
jaxb-api
${jaxb-api.version}
```
+ Il jar del connettore Spark Kinesis si trova in questo percorso dopo la creazione di un cluster EMR: `/usr/lib/spark/connector/lib/`
# S3 (s3-dist-cp) DistCp
Apache DistCp è uno strumento open source che puoi utilizzare per copiare grandi quantità di dati. *S3 DistCp* è simile DistCp, ma ottimizzato per funzionare AWS, in particolare con Amazon S3. Il comando per S3 DistCp in Amazon EMR versione 4.0 e successive `s3-dist-cp` è quello che aggiungi come passaggio in un cluster o nella riga di comando. Utilizzando S3DistCp, puoi copiare in modo efficiente grandi quantità di dati da Amazon S3 in HDFS, dove possono essere elaborati con passaggi successivi nel tuo cluster Amazon EMR. Puoi anche usare S3 DistCp per copiare dati tra bucket Amazon S3 o da HDFS ad Amazon S3. S3 DistCp è più scalabile ed efficiente per la copia parallela di un gran numero di oggetti tra bucket e account. AWS
Per i comandi specifici che dimostrano la flessibilità di S3DistCP in scenari reali, consulta [Sette suggerimenti](https://aws.amazon.com/blogs/big-data/seven-tips-for-using-s3distcp-on-amazon-emr-to-move-data-efficiently-between-hdfs-and-amazon-s3/) per l'utilizzo di S3 sul blog Big Data. DistCp AWS
Ad esempio DistCp, S3 DistCp utilizza MapReduce la copia in modo distribuito. Condivide la copia, la gestione degli errori, il ripristino e le attività di creazione di report su più server. Per ulteriori informazioni sul progetto DistCp open source Apache, consulta la [DistCpguida](http://hadoop.apache.org/docs/stable/hadoop-distcp/DistCp.html) nella documentazione di Apache Hadoop.
Se S3 non DistCp è in grado di copiare alcuni o tutti i file specificati, la fase del cluster fallisce e restituisce un codice di errore diverso da zero. In questo caso, S3 non DistCp pulisce i file parzialmente copiati.
**Importante**
S3 non DistCp supporta i nomi di bucket Amazon S3 che contengono il carattere di sottolineatura.
S3 DistCp non supporta la concatenazione per i file Parquet. Usa invece. PySpark Per ulteriori informazioni, consulta la sezione relativa alla [Concatenazione dei file Parquet in Amazon EMR](https://aws.amazon.com/premiumsupport/knowledge-center/emr-concatenate-parquet-files/).
Per evitare errori di copia quando si utilizza S3DistCP per copiare un singolo file (anziché una directory) da S3 a HDFS, utilizza Amazon EMR versione 5.33.0 o successive o Amazon EMR versione 6.3.0 o successive.
## Opzioni S3 DistCp
Sebbene sia simile a DistCp, S3 DistCp supporta un diverso set di opzioni per modificare il modo in cui copia e comprime i dati.
Quando chiami S3DistCp, puoi specificare le opzioni descritte nella tabella seguente. Le opzioni vengono aggiunte alla fase con l'elenco di argomenti. Nella tabella seguente sono riportati alcuni esempi DistCp degli argomenti S3.
| Opzione | Description | Richiesto |
| --- | --- | --- |
| ‑‑src=LOCATION | Posizione dei dati da copiare. Può essere una posizione HDFS o Amazon S3. Ad esempio: `‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node` S3 non DistCp supporta i nomi di bucket Amazon S3 che contengono il carattere di sottolineatura. | Sì |
| ‑‑dest=LOCATION | Destinazione per i dati. Può essere una posizione HDFS o Amazon S3. Ad esempio: `‑‑dest=hdfs:///output` S3 non DistCp supporta i nomi di bucket Amazon S3 che contengono il carattere di sottolineatura. | Sì |
| ‑‑srcPattern=PATTERN | Un'[espressione regolare](http://en.wikipedia.org/wiki/Regular_expression) che filtra l'operazione di copia per un sottoinsieme dei dati in `‑‑src`. Se non è specificato né `‑‑srcPattern` né `‑‑groupBy`, tutti i dati in `‑‑src` vengono copiati su `‑‑dest`. Se l'argomento dell'espressione regolare contiene caratteri speciali quali l'asterisco (\$1), l'espressione regolare o l'intera stringa `‑‑args` deve essere racchiusa tra virgolette singole ('). Ad esempio: `‑‑srcPattern=.*daemons.*-hadoop-.*` | No |
| ‑‑groupBy=PATTERN | Un'[espressione regolare](http://en.wikipedia.org/wiki/Regular_expression) che fa sì che S3 concateni i file che DistCp corrispondono all'espressione. Ad esempio, è possibile utilizzare questa opzione per riunire in un unico file tutti i file di log scritti in un'ora. Il filename concatenato è il valore trovato dall'espressione regolare per il raggruppamento. Le parentesi indicano il modo in cui i file devono essere raggruppati, con tutti gli elementi che soddisfano l'istruzione tra parentesi riuniti in un singolo file di output. Se l'espressione regolare non include un'istruzione tra parentesi, il cluster ha esito negativo nella fase S3 e restituisce un errore. DistCp Se l'argomento dell'espressione regolare contiene caratteri speciali quali l'asterisco (\$1), l'espressione regolare o l'intera stringa `‑‑args` deve essere racchiusa tra virgolette singole ('). Se è specificato `‑‑groupBy`, vengono copiati solo i file che corrispondono al modello specificato. Non è necessario specificare `‑‑groupBy` e `‑‑srcPattern` contemporaneamente. Ad esempio: `‑‑groupBy=.*subnetid.*([0-9]+-[0-9]+-[0-9]+-[0-9]+).*` | No |
| ‑‑targetSize=SIZE | La dimensione in mebibyte (MiB) dei file da creare in base all'opzione `‑‑groupBy`. Questo valore deve essere un numero intero. Quando `‑‑targetSize` è impostata, S3 DistCp tenta di corrispondere a questa dimensione; la dimensione effettiva dei file copiati può essere maggiore o minore di questo valore. I processi vengono aggregati in base alle dimensioni del file di dati, pertanto è possibile che la dimensione del file di destinazione corrisponda a quella del file di origine. Se i file concatenati da `‑‑groupBy` hanno una dimensione maggiore rispetto al valore di `‑‑targetSize`, vengono suddivisi in file part e denominati in sequenza con un valore numerico aggiunto alla fine. Ad esempio, un file concatenato in `myfile.gz` sarebbe suddiviso in parti come: `myfile0.gz`, `myfile1.gz` ecc. Ad esempio: `‑‑targetSize=2` | No |
| ‑‑appendToLastFile | Specifica il comportamento di S3 DistCp durante la copia di file da Amazon S3 a HDFS già presenti. Aggiunge i nuovi dati ai file esistenti. Se utilizzi `‑‑appendToLastFile` con `‑‑groupBy`, i nuovi dati vengono aggiunti ai file che corrispondono agli stessi gruppi. Questa opzione rispetta anche il comportamento `‑‑targetSize` quando usato con `‑‑groupBy.` | No |
| ‑‑outputCodec=CODEC | Specifica i codec di compressione da utilizzare per i file copiati. I valori possibili sono `gzip`, `gz`, `lzo`, `snappy` o `none`. È possibile utilizzare questa opzione, ad esempio, per convertire i file di input compressi con Gzip in file di output con compressione LZO o per decomprimere i file durante l'operazione di copia. Se scegli un codec di output, il filename viene aggiunto con l'estensione appropriata (ad esempio per `gz` e `gzip`, l'estensione è `.gz`). Se non specifichi un valore per `‑‑outputCodec`, i file vengono copiati senza apportare modifiche alla compressione. Ad esempio: `‑‑outputCodec=lzo` | No |
| ‑‑s3ServerSideEncryption | Assicura che i dati di destinazione vengano trasferiti tramite SSL e crittografati automaticamente in Amazon S3 utilizzando AWS una chiave lato servizio. Quando si recuperano dati utilizzando S3, gli oggetti vengono automaticamente DistCp decrittografati. Se tenti di copiare un oggetto non crittografato in un bucket Amazon S3 che richiede la crittografia, l'operazione ha esito negativo. Per maggiori informazioni, consulta [Utilizzo della crittografia dei dati](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingEncryption.html). Ad esempio: `‑‑s3ServerSideEncryption` | No |
| ‑‑deleteOnSuccess | Se l'operazione di copia ha esito positivo, questa opzione fa sì che S3 elimini DistCp i file copiati dalla posizione di origine. Questa funzione è utile se stai copiando file di output, ad esempio i file di log, da una posizione a un'altra come attività pianificata e non vuoi copiare lo stesso file due volte. Ad esempio: `‑‑deleteOnSuccess` | No |
| ‑‑disableMultipartUpload | Disattiva l'utilizzo del caricamento in più parti. Ad esempio: `‑‑disableMultipartUpload` | No |
| ‑‑multipartUploadChunkSize=SIZE | Le dimensioni, in MiB, di ogni parte in un caricamento in più parti di Amazon S3. S3 DistCp utilizza il caricamento in più parti quando copia dati di dimensioni superiori a. `multipartUploadChunkSize` Per migliorare le prestazioni del processo, puoi aumentare le dimensioni di ogni parte. La dimensione predefinita è 128 MiB. Ad esempio: `‑‑multipartUploadChunkSize=1000` | No |
| ‑‑numberFiles | Antepone numeri sequenziali ai file di output. Il conteggio inizia da 0, a meno che non venga specificato un valore diverso tramite `‑‑startingIndex`. Ad esempio: `‑‑numberFiles` | No |
| ‑‑startingIndex=INDEX | Utilizzato con `‑‑numberFiles` per specificare il primo numero nella sequenza. Ad esempio: `‑‑startingIndex=1` | No |
| ‑‑outputManifest=FILENAME | Crea un file di testo, compresso con Gzip, che contiene un elenco di tutti i file copiati da S3. DistCp Ad esempio: `‑‑outputManifest=manifest-1.gz` | No |
| ‑‑previousManifest=PATH | Legge un file manifest creato durante una precedente chiamata a S3 utilizzando il flag. DistCp `‑‑outputManifest` Quando il `‑‑previousManifest` flag è impostato, S3 DistCp esclude i file elencati nel manifesto dall'operazione di copia. Se viene specificato `‑‑outputManifest` insieme a `‑‑previousManifest`, i file elencati nel manifest precedente compaiono anche nel nuovo file manifest, pur non venendo copiati. Ad esempio: `‑‑previousManifest=/usr/bin/manifest-1.gz` | No |
| ‑‑requirePreviousManifest | Richiede un manifesto precedente creato durante una precedente chiamata a S3. DistCp Se impostato su false, non vengono generati errori se non viene specificato un manifest precedente. Il valore predefinito è true. | No |
| ‑‑copyFromManifest | Inverte il comportamento di `‑‑previousManifest` far sì che S3 DistCp utilizzi il file manifest specificato come elenco di file da copiare, anziché come elenco di file da escludere dalla copia. Ad esempio: `‑‑copyFromManifest ‑‑previousManifest=/usr/bin/manifest-1.gz` | No |
| ‑‑s3Endpoint=ENDPOINT | Specifica l'endpoint Amazon S3 da utilizzare durante il caricamento di un file. Questa opzione imposta l'endpoint sia per l'origine che per la destinazione. Se non è impostato, l'endpoint predefinito è `s3.amazonaws.com`. Per un elenco degli endpoint di Amazon S3, consulta la sezione relativa a [Regioni ed endpoint](https://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region). Ad esempio: `‑‑s3Endpoint=s3.eu-west-1.amazonaws.com` | No |
| ‑‑storageClass=CLASS | La classe di archiviazione da utilizzare quando la destinazione è Amazon S3. I valori validi sono STANDARD e REDUCED\$1REDUNDANCY. Se questa opzione non è specificata, S3 DistCp tenta di preservare la classe di archiviazione. Ad esempio: `‑‑storageClass=STANDARD` | No |
| ‑‑srcPrefixesFile=PATH | Un file di testo in Amazon S3 (s3://), HDFS (hdfs:///) o nel file system locale (file:/) che contiene un elenco di prefissi `src`, uno per riga. Se `srcPrefixesFile` viene fornito, S3 non DistCp elencherà il percorso src. Al contrario, genera un elenco di origine combinando tutti gli elenchi di prefissi specificati in questo file. Per generare percorsi di destinazione, invece dei prefissi verrà utilizzato il percorso relativo rispetto al percorso di src. Se viene specificato anche `srcPattern`, verrà applicato ai risultati dell'elenco combinato dei prefissi di origine per filtrare ulteriormente l'input. Se viene utilizzato `copyFromManifest`, gli oggetti nel manifest verranno copiati e `srcPrefixesFile` verrà ignorato. Ad esempio: `‑‑srcPrefixesFile=PATH` | No |
Oltre alle opzioni precedenti, S3 DistCp implementa l'[interfaccia Tool](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/util/Tool.html), il che significa che supporta le opzioni generiche.
## Aggiungere S3 DistCp come fase in un cluster
Puoi chiamare S3 DistCp aggiungendolo come passaggio nel cluster. È possibile aggiungere fasi a un cluster all'avvio o in esecuzione utilizzando la console, la CLI o l'API. Gli esempi seguenti mostrano l'aggiunta di un DistCp passaggio S3 a un cluster in esecuzione. Per ulteriori informazioni sull'aggiunta di fasi a un cluster, consulta [Invio di lavoro a un cluster](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-work-with-steps.html) nella *Guida alla gestione di Amazon EMR*.
**Per aggiungere un DistCp passaggio S3 a un cluster in esecuzione utilizzando il AWS CLI**
Per ulteriori informazioni sull'utilizzo dei comandi Amazon EMR in AWS CLI, consulta il [AWS CLI Command](https://docs.aws.amazon.com/cli/latest/reference/emr) Reference.
+ Per aggiungere un passaggio a un cluster che chiama S3DistCp, inserisci come argomenti i parametri che specificano come S3 DistCp deve eseguire l'operazione di copia.
L'esempio seguente copia il log dei daemon da Amazon S3 a `hdfs:///output`. Nel comando riportato qui di seguito:
+ `‑‑cluster-id` specifica il cluster
+ `Jar`è la posizione del file JAR S3DistCp . Per un esempio di come eseguire un comando su un cluster utilizzando command-runner.jar, consulta [Invia una fase JAR personalizzata per l'esecuzione di uno script o di un comando](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-commandrunner.html#emr-commandrunner-examples).
+ `Args`è un elenco separato da virgole delle coppie nome-valore di opzioni da passare a S3. DistCp Per un elenco completo delle opzioni disponibili, consulta [Opzioni S3 DistCp](#UsingEMR_s3distcp.options).
Per aggiungere una fase di DistCp copia S3 a un cluster in esecuzione, inserisci quanto segue in un file JSON salvato in Amazon S3 o nel tuo file system locale, `myStep.json` come in questo esempio. *j-3GYXXXXXX9IOK*Sostituiscilo con il tuo ID cluster e *amzn-s3-demo-bucket* sostituiscilo con il nome del bucket Amazon S3.
```
[
{
"Name":"S3DistCp step",
"Args":["s3-dist-cp","‑‑s3Endpoint=s3.amazonaws.com","‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node/","‑‑dest=hdfs:///output","‑‑srcPattern=.*[a-zA-Z,]+"],
"ActionOnFailure":"CONTINUE",
"Type":"CUSTOM_JAR",
"Jar":"command-runner.jar"
}
]
```
```
aws emr add-steps ‑‑cluster-id j-3GYXXXXXX9IOK ‑‑steps file://./myStep.json
```
**Example Copia dei file di log da Amazon S3 ad HDFS**
Questo esempio illustra anche come copiare in HDFS i file di log archiviati in un bucket Amazon S3 aggiungendo una fase a un cluster in esecuzione. In questo esempio l'opzione `‑‑srcPattern` viene utilizzata per limitare i dati copiati nel log dei daemon.
Per copiare i file di log da Amazon S3 a HDFS utilizzando l'opzione `‑‑srcPattern`, inserisci quanto segue in un file JSON salvato in Amazon S3 o nel file system locale come `myStep.json` per questo esempio. *j-3GYXXXXXX9IOK*Sostituiscilo con il tuo ID cluster e *amzn-s3-demo-bucket* sostituiscilo con il nome del bucket Amazon S3.
```
[
{
"Name":"S3DistCp step",
"Args":["s3-dist-cp","‑‑s3Endpoint=s3.amazonaws.com","‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node/","‑‑dest=hdfs:///output","‑‑srcPattern=.*daemons.*-hadoop-.*"],
"ActionOnFailure":"CONTINUE",
"Type":"CUSTOM_JAR",
"Jar":"command-runner.jar"
}
]
```
## Pulizia dopo lavori S3 non riusciti DistCp
Se S3 DistCp non è in grado di copiare alcuni o tutti i file specificati, il comando o la fase del cluster fallisce e restituisce un codice di errore diverso da zero. In questo caso, S3 non DistCp pulisce i file parzialmente copiati. È necessario eliminarli manualmente.
I file parzialmente copiati vengono salvati nella directory HDFS in `tmp` sottodirectory con l'identificatore univoco del lavoro S3. DistCp È possibile trovare questo ID nell'output standard dell'attività.
Ad esempio, per un DistCp lavoro S3 con l'ID`4b1c37bb-91af-4391-aaf8-46a6067085a6`, è possibile connettersi al nodo principale del cluster ed eseguire il comando seguente per visualizzare i file di output associati al lavoro.
```
hdfs dfs -ls /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output
```
Il comando restituisce un elenco di file simile al seguente:
```
Found 8 items
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/_SUCCESS
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00000
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00001
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00002
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00003
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00004
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00005
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00006
```
È quindi possibile eseguire il seguente comando per eliminare la directory e tutti i relativi contenuti.
```
hdfs dfs rm -rf /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6
```