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à. # Migliorare le prestazioni di Spark con Amazon S3 Amazon EMR offre caratteristiche che aiutano a ottimizzare le prestazioni quando si usa Spark per eseguire query e leggere e scrivere i dati salvati in Amazon S3. [S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) può migliorare le prestazioni delle query per i file CSV e JSON in alcune applicazioni mediante il pushdown dell'elaborazione ad Amazon S3. Il committer ottimizzato per EMRFS S3 è un'alternativa alla [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html)classe, che utilizza la funzionalità di caricamento multiparte di EMRFS per migliorare le prestazioni durante la scrittura di file Parquet su Amazon S3 utilizzando Spark e Datasets. DataFrames **Topics** + [Usare S3 Select con Spark per migliorare le prestazioni delle query](emr-spark-s3select.md) + [EMR Spark MagicCommitProtocol](emr-spark-magic-commit-protocol.md) + [Utilizzare il committer ottimizzato S3 EMRFS](emr-spark-s3-optimized-committer.md) + [Utilizzo del protocollo di commit ottimizzato per S3 EMRFS](emr-spark-s3-optimized-commit-protocol.md) + [Effettuare nuovi tentativi per le richieste Amazon S3 con EMRFS](emr-spark-emrfs-retry.md) # Usare S3 Select con Spark per migliorare le prestazioni delle query **Importante** Amazon S3 Select non è più disponibile per i nuovi clienti. I clienti esistenti di Amazon S3 Select possono continuare a utilizzare la funzionalità come di consueto. [Ulteriori informazioni](https://aws.amazon.com/blogs/storage/how-to-optimize-querying-your-data-in-amazon-s3/) Con Amazon EMR rilascio 5.17.0 e successivi, puoi utilizzare [S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) con Spark su Amazon EMR. *S3 Select* consente alle applicazioni il recupero di un solo sottoinsieme di dati da un oggetto. Per Amazon EMR, l'attività di calcolo di filtro di set di dati di grandi dimensioni per l'elaborazione viene "trasferita" dal cluster ad Amazon S3; in tal modo è possibile migliorare le prestazioni in alcune applicazioni e ridurre la quantità di dati trasferiti tra Amazon EMR e Amazon S3. S3 Select è supportata con file CSV e JSON che utilizzano i valori `s3selectCSV` e `s3selectJSON` per specificare il formato dei dati. Per maggiori informazioni ed esempi, consulta [Specificare S3 Select nel codice](#emr-spark-s3select-specify). ## S3 Select è adatto alla mia applicazione? È consigliabile eseguire analisi comparative delle applicazioni con e senza S3 Select per vedere se l'utilizzo di tale caratteristica può risultare idoneo per l'applicazione. Per determinare se l'applicazione è idonea per l'uso di S3 Select, utilizza le seguenti linee guida: + La query filtra più di metà del set di dati originale. + La connessione di rete tra Amazon S3 e il cluster Amazon EMR ha una buona velocità di trasferimento e larghezza di banda disponibile. Amazon S3 non comprime le risposte HTTP, pertanto è probabile che le dimensioni della risposta aumentino per file di input compressi. ## Considerazioni e limitazioni + La crittografia lato server con chiavi di crittografia fornite dal cliente (SSE-C) e la crittografia lato client di Amazon S3 non sono supportate. + La proprietà `AllowQuotedRecordDelimiters` non è supportata. Se questa proprietà è specificata, la query ha esito negativo. + Sono supportati solo i file CSV e JSON in formato UTF-8. Le CSVs linee multiple non sono supportate. + Sono supportati solo i file non compressi o gzip. + Le opzioni Spark CSV e JSON quali `nanValue`, `positiveInf`, `negativeInf` e le opzioni relative a record danneggiati (ad esempio, le modalità failfast e dropmalformed) non sono supportate. + L'utilizzo di virgole (,) all'interno di decimali non è supportato. Ad esempio, `10,000` non è supportato, mentre è supportato `10000`. + I caratteri di commento nell'ultima riga non sono supportati. + Le righe vuote alla fine di un file non vengono elaborate. + I filtri seguenti non sono trasferiti ad Amazon S3: + Funzioni di aggregazione quali `COUNT()` e `SUM()`. + Filtri che eseguono il `CAST()` di un attributo. Ad esempio, `CAST(stringColumn as INT) = 1`. + Filtri con un attributo che è un oggetto o è complesso. Ad esempio, `intArray[1] = 1, objectColumn.objectNumber = 1`. + I filtri per i quali il valore non è un valore letterale. Ad esempio, `intColumn1 = intColumn2` + Sono supportati solo [tipi di dati supportati da S3 Select](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-glacier-select-sql-reference-data-types.html) con le limitazioni documentate. ## Specificare S3 Select nel codice Gli esempi seguenti mostrano come specificare S3 Select per CSV utilizzando Scala, SQL, R e. PySpark È possibile utilizzare S3 Select per JSON nello stesso modo. Per un elenco delle opzioni, dei relativi valori predefiniti e le limitazioni, consulta [Opzioni](#emr-spark-s3select-specify-options). ------ #### [ PySpark ] ``` spark .read .format("s3selectCSV") // "s3selectJson" for Json .schema(...) // optional, but recommended .options(...) // optional .load("s3://path/to/my/datafiles") ``` ------ #### [ R ] ``` read.df("s3://path/to/my/datafiles", "s3selectCSV", schema, header = "true", delimiter = "\t") ``` ------ #### [ Scala ] ``` spark .read .format("s3selectCSV") // "s3selectJson" for Json .schema(...) // optional, but recommended .options(...) // optional. Examples: // .options(Map("quote" -> "\'", "header" -> "true")) or // .option("quote", "\'").option("header", "true") .load("s3://path/to/my/datafiles") ``` ------ #### [ SQL ] ``` CREATE TEMPORARY VIEW MyView (number INT, name STRING) USING s3selectCSV OPTIONS (path "s3://path/to/my/datafiles", header "true", delimiter "\t") ``` ------ ### Opzioni Le seguenti opzioni sono disponibili quando si utilizzano `s3selectCSV` e `s3selectJSON`. Se non specificate, verranno utilizzati i valori predefiniti. #### Opzioni con S3selectCSV | Opzione | Predefinita | Utilizzo | | --- | --- | --- | | `compression` | `"none"` | Indica se viene utilizzata la compressione. `"gzip"` è l'unica impostazione supportata oltre a `"none"`. | | `delimiter` | "," | Specifica il delimitatore del campo. | | `quote` | `'\"'` | Specifica il carattere virgoletta. La specifica di una stringa vuota non è supportata e genera un errore di XML danneggiato. | | `escape` | `'\\'` | Specifica il carattere di escape. | | `header` | `"false"` | `"false"` specifica che non è presente alcuna intestazione. `"true"` specifica che nella prima riga è presente un'intestazione. Sono supportate solo le intestazioni nella prima riga e non sono supportate righe vuote prima di un'intestazione. | | comment | `"#"` | Specifica il carattere per commenti. Non è possibile disabilitare l'indicatore dei commenti. In altre parole, il valore `\u0000` non è supportato. | | `nullValue` | "" | | #### Opzioni con S3selectJSON | Opzione | Predefinita | Utilizzo | | --- | --- | --- | | `compression` | `"none"` | Indica se viene utilizzata la compressione. `"gzip"` è l'unica impostazione supportata oltre a `"none"`. | | `multiline` | "false" | `"false"` specifica che JSON è nel formato `LINES` S3 Select, che significa che ogni riga dei dati di input contiene un singolo oggetto JSON. `"true"` specifica che JSON è nel formato `DOCUMENT` S3 Select, che significa che un oggetto JSON può estendersi su più righe dei dati di input. | # EMR Spark MagicCommitProtocol A partire da EMR 6.15.0, MagicCommitProtocol diventa l'impostazione predefinita FileCommitProtocol per Spark quando si utilizza il file system S3A. ## MagicCommitProtocol MagicCommitProtocol Si tratta di un'implementazione alternativa ottimizzata per [FileCommitProtocol](https://dlcdn.apache.org/spark/docs/2.4.2/api/java/org/apache/spark/internal/io/FileCommitProtocol.html)la scrittura di file con EMR Spark su Amazon S3 quando si utilizza il file system S3A. Questo protocollo mira a migliorare le prestazioni delle applicazioni evitando l'uso di operazioni di ridenominazione in Amazon S3 durante le fasi di job e task commit. MagicCommitProtocol È l' FileCommitProtocol implementazione predefinita utilizzata da Spark in esecuzione su Amazon Elastic Map Reduce (EMR) quando viene utilizzato il file system S3A. Utilizza MagicCommitProtocol internamente [MagicV2Committer](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/s3a-magicv2-committer.html) per eseguire le scritture di file su Amazon S3. Per le operazioni di inserimento statiche, MagicCommitProtocol scrive i file nella posizione di output del lavoro durante la fase di conferimento dell'attività. Al contrario, per le operazioni dinamiche di sovrascrittura degli inserti, i file scritti dai tentativi di operazione vengono visualizzati nella posizione di output del lavoro solo dopo il completamento del lavoro. Ciò si ottiene esportando nuovamente i metadati di commit nel driver Spark durante la chiamata di commit del task. ## Abilitazione MagicCommitProtocol MagicCommitProtocol È abilitato per impostazione predefinita per Spark in esecuzione su Amazon Elastic Map Reduce (EMR) quando si utilizza il file system S3A. Per utilizzare il file system S3A, puoi: 1. Usa lo schema di file come `s3a://` quando definisci la tabella, la partizione o la directory. 1. Imposta la configurazione `fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem` in core-site.xml. ## Disabilitazione di MagicCommitProtocol 1. Puoi `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` impostarlo su false codificandolo in a`SparkConf`, passandolo come `--conf` parametro nella shell Spark o `spark-submit` and `spark-sql` tools, oppure in. `conf/spark-defaults.conf` Per ulteriori informazioni, consulta la [configurazione di Spark nella documentazione di Apache Spark](https://spark.apache.org/docs/latest/configuration.html). L'esempio seguente mostra come disabilitare MagicCommitProtocol durante l'esecuzione di un comando. `spark-sql` ``` spark-sql \ --conf spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol=false \ -e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;" ``` 1. Utilizzate la classificazione di `spark-defaults` configurazione per impostare la `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` proprietà su false. Per ulteriori informazioni, consulta la sezione [Configurazione delle applicazioni](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html). ## MagicCommitProtocol considerazioni + Per l'inserimento di partizioni statiche, negli esecutori Spark, MagicCommitProtocol consuma una piccola quantità di memoria per ogni file scritto da un tentativo di operazione fino a quando l'operazione non viene completata o interrotta. Nella maggior parte dei processi la quantità di memoria consumata è trascurabile. Il driver Spark non richiede memoria aggiuntiva + Per l'inserimento dinamico delle partizioni, sui driver Spark, è MagicCommitProtocol necessaria memoria per archiviare le informazioni sui metadati di ogni file salvato fino al completamento o all'interruzione del lavoro. Nella maggior parte dei processi, l'impostazione predefinita della memoria del driver Spark è trascurabile. Per i processi che presentano attività di lunga esecuzione che scrivono un numero elevato di file, la memoria che il protocollo di commit consuma può essere notevole e rende necessario adeguare la memoria allocata per Spark, specialmente per gli esecutori Spark. È possibile regolare la memoria utilizzando la proprietà `spark.driver.memory` per i driver Spark e la proprietà `spark.executor.memory` per gli esecutori Spark. Come linea guida, una singola operazione di scrittura di 100.000 file richiederebbe in genere 200 MB di memoria aggiuntivi. Per ulteriori informazioni, consulta le [Proprietà delle applicazioni](https://spark.apache.org/docs/latest/configuration.html#application-properties) nella documentazione relativa alla configurazione di Apache Spark. # Utilizzare il committer ottimizzato S3 EMRFS Il committer ottimizzato per EMRFS S3 è un'[OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html)implementazione alternativa ottimizzata per la scrittura di file su Amazon S3 quando si utilizza EMRFS. Il committer ottimizzato per S3 EMRFS migliora le prestazioni delle applicazioni evitando l'esecuzione di operazioni di elenchi e di ridenominazione in Amazon S3 durante le fasi di commit di processi e attività. Il committer è disponibile in Amazon EMR rilascio 5.19.0 e successivi ed è abilitato per impostazione predefinita in Amazon EMR rilascio 5.20.0 e successivi. Il committer viene utilizzato per i job Spark che utilizzano Spark o Datasets. DataFrames A partire da Amazon EMR 6.4.0, questo committer può essere utilizzato per tutti i formati comuni, tra cui Parquet, ORC e formati testuali (inclusi CSV e JSON). Per i rilasci precedenti ad Amazon EMR 6.4.0, è supportato solo il formato Parquet. Vi sono circostanze in cui il committer non viene utilizzato. Per ulteriori informazioni, consulta [Requisiti per il committer ottimizzato S3 EMRFS](emr-spark-committer-reqs.md). **Topics** + [Requisiti per il committer ottimizzato S3 EMRFS](emr-spark-committer-reqs.md) + [Committer ottimizzato S3 EMRFS e caricamenti in più parti](emr-spark-committer-multipart.md) + [Considerazioni sull'ottimizzazione dei processi](emr-spark-committer-tuning.md) + [Abilitare il committer ottimizzato S3 EMRFS per Amazon EMR 5.19.0](emr-spark-committer-enable.md) # Requisiti per il committer ottimizzato S3 EMRFS Il committer ottimizzato S3 EMRFS viene utilizzato quando si verificano le condizioni riportate di seguito: + Esegui job Spark che utilizzano Spark o Datasets per scrivere file su Amazon S3. DataFrames A partire da Amazon EMR 6.4.0, questo committer può essere utilizzato per tutti i formati comuni, tra cui Parquet, ORC e formati testuali (inclusi CSV e JSON). Per i rilasci precedenti ad Amazon EMR 6.4.0, è supportato solo il formato Parquet. + I caricamenti in più parti sono abilitati in Amazon EMR. Questa è l’impostazione predefinita. Per ulteriori informazioni, consulta [Committer ottimizzato S3 EMRFS e caricamenti in più parti](emr-spark-committer-multipart.md). + Viene utilizzato il supporto integrato per i formati di file di Spark. Il supporto integrato per i formati viene utilizzato nelle seguenti circostanze: + Per le tabelle Metastore Hive, quando `spark.sql.hive.convertMetastoreParquet` è impostato su `true` per le tabelle Parquet, oppure quando `spark.sql.hive.convertMetastoreOrc` è impostato su `true` per le tabelle Orc con Amazon EMR versioni 6.4.0 o successive. Queste sono le impostazioni predefinite. + Quando i processi scrivono in origini dei dati o tabelle di formato di file, ad esempio la tabella di destinazione viene creata con la clausola `USING parquet`. + Quando i processi scrivono tabelle Parquet non partizionate Hive metastore. Il supporto per Parquet integrato di Spark non supporta le tabelle Hive partizionate, che è una limitazione nota. Per ulteriori informazioni, consulta [Hive metastore Parquet table conversion](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#hive-metastore-parquet-table-conversion) nella Apache Spark and Datasets Guide. DataFrames + Le operazioni di processo Spark che scrivono in una posizione di partizione predefinita, ad esempio `${table_location}/k1=v1/k2=v2/`, usano il committer. Il committer non viene utilizzato se un'operazione di processo scrive in una posizione di partizione personalizzata, ad esempio se un percorso di partizione personalizzato è impostato utilizzando il comando `ALTER TABLE SQL`. + Devono essere utilizzati i seguenti valori per Spark: + La proprietà `spark.sql.parquet.fs.optimized.committer.optimization-enabled` deve essere impostata su `true`. Questa è l'impostazione predefinita con Amazon EMR 5.20.0 e versione successiva. Con Amazon EMR 5.19.0, il valore predefinito è `false`. Per informazioni su come configurare questo valore, consulta [Abilitare il committer ottimizzato S3 EMRFS per Amazon EMR 5.19.0](emr-spark-committer-enable.md). + Se si scrive su tabelle metastore Hive non partizionate, sono supportati solo i formati di file Parquet e Orc. `spark.sql.hive.convertMetastoreParquet`deve essere impostato su `true` se si scrive su tabelle metastore di Parquet Hive non partizionate. `spark.sql.hive.convertMetastoreOrc`deve essere impostato su `true` se si scrive su tabelle metastore di Orc Hive non partizionate. Queste sono le impostazioni predefinite. + `spark.sql.parquet.output.committer.class` deve essere impostato su `com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter`. Questa è l'impostazione predefinita. + `spark.sql.sources.commitProtocolClass` deve essere impostato su `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` o `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol`. `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` è l'impostazione predefinita per la serie Amazon EMR 5.x versione 5.30.0 e successive e per la serie Amazon EMR 6.x versione 6.2.0 e successive. `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` è l'impostazione predefinita per le versioni Amazon EMR precedenti. + Se i processi Spark sovrascrivono i set di dati Parquet partizionati con colonne di partizione dinamiche, le opzioni di scrittura `partitionOverwriteMode` e `spark.sql.sources.partitionOverwriteMode` devono essere impostate su `static`. Questa è l'impostazione predefinita. **Nota** L'opzione di scrittura `partitionOverwriteMode` è stata introdotta in Spark 2.4.0. Per Spark versione 2.3.2, incluso con Amazon EMR rilascio 5.19.0, imposta la proprietà `spark.sql.sources.partitionOverwriteMode`. ## Quando il committer ottimizzato S3 EMRFS non viene utilizzato In genere, il committer EMRFS ottimizzato per S3 EMRFS non viene utilizzato nelle situazioni riportate di seguito. **** | Situazione | Perché il committer non viene utilizzato | | --- | --- | | Quando si scrive su HDFS | Il committer supporta solo la scrittura su Amazon S3 utilizzando EMRFS. | | Quando si utilizza il file system S3A | Il committer supporta solo EMRFS. | | Quando utilizzi l'API RDD di Spark MapReduce | Il committer supporta solo l'utilizzo di SparkSQL o DataFrame Dataset. APIs | I seguenti esempi di Scala mostrano altre situazioni che impediscono di utilizzare il committer EMRFS ottimizzato per S3 in tutto (il primo esempio) e in parte (il secondo esempio). **Example - Modalità di sovrascrittura dinamica delle partizioni** Il seguente esempio di Scala indica a Spark di utilizzare un algoritmo di commit diverso, che impedisce l'uso del committer EMRFS ottimizzato per S3. Il codice imposta la proprietà `partitionOverwriteMode` su `dynamic` per sovrascrivere solo le partizioni su cui si stanno scrivendo i dati. Quindi, le colonne delle partizioni dinamiche vengono specificate da `partitionBy` e la modalità scrittura è impostata su `overwrite`. ``` val dataset = spark.range(0, 10) .withColumn("dt", expr("date_sub(current_date(), id)")) dataset.write.mode("overwrite") .option("partitionOverwriteMode", "dynamic") .partitionBy("dt") .parquet("s3://amzn-s3-demo-bucket1/output") ``` È necessario configurare tutte e tre le impostazioni per evitare l'utilizzo del committer EMRFS ottimizzato per S3. In questo modo, Spark esegue un algoritmo di commit diverso specificato nel protocollo di commit di Spark. Per i rilasci di Amazon EMR serie 5.x precedenti al 5.30.0 e per i rilasci di Amazon EMR serie 6.x precedenti a 6.2.0, il protocollo di commit utilizza la directory di gestione temporanea di Spark, che è una directory temporanea creata nella posizione di output che inizia con `.spark-staging`. L'algoritmo rinomina in modo sequenziale le directory delle partizioni e questo può influire negativamente sulle prestazioni. Per ulteriori informazioni su Amazon EMR rilascio 5.30.0 e successivi e sul rilascio 6.2.0 e successivi, consulta la sezione [Utilizzo del protocollo di commit ottimizzato per S3 EMRFS](emr-spark-s3-optimized-commit-protocol.md). L'algoritmo in Spark 2.4.0 segue questi passaggi: 1. I tentativi di attività scrivono il loro output nelle directory delle partizioni sotto la directory di staging di Spark, ad esempio `${outputLocation}/spark-staging-${jobID}/k1=v1/k2=v2/`. 1. Per ogni partizione scritta, il tentativo di attività tiene traccia dei percorsi di partizione relativi, ad esempio `k1=v1/k2=v2`. 1. Quando un'attività viene completata correttamente, fornisce al driver tutti i relativi percorsi di partizione che ha tracciato. 1. Al termine di tutte le attività, la fase di commit dei lavori raccoglie tutte le directory delle partizioni che i tentativi di attività riusciti hanno scritto nella directory di gestione temporanea di Spark. Spark rinomina in sequenza ciascuna di queste directory nella sua posizione di output finale utilizzando le operazioni di ridenominazione dell'albero delle directory. 1. La directory di gestione temporanea viene eliminata prima del completamento della fase di commit del processo. **Example - Posizione della partizione personalizzata** In questo esempio, il codice Scala viene inserito in due partizioni. Una partizione ha una posizione di partizione personalizzata. L'altra partizione usa il percorso di partizione predefinito. Il committer ottimizzato S3 EMRFS viene utilizzato solo per scrivere l'output dell'attività nella partizione che utilizza la posizione della partizione predefinita. ``` val table = "dataset" val location = "s3://bucket/table" spark.sql(s""" CREATE TABLE $table (id bigint, dt date) USING PARQUET PARTITIONED BY (dt) LOCATION '$location' """) // Add a partition using a custom location val customPartitionLocation = "s3://bucket/custom" spark.sql(s""" ALTER TABLE $table ADD PARTITION (dt='2019-01-28') LOCATION '$customPartitionLocation' """) // Add another partition using default location spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')") def asDate(text: String) = lit(text).cast("date") spark.range(0, 10) .withColumn("dt", when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29"))) .write.insertInto(table) ``` Il codice Scala crea i seguenti oggetti Amazon S3: ``` custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet custom_$folder$ table/_SUCCESS table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet table/dt=2019-01-29_$folder$ table_$folder$ ``` Quando si scrive nelle partizioni in posizioni personalizzate, Spark utilizza un algoritmo di commit simile all'esempio precedente, che è descritto di seguito. Come nell'esempio precedente, l'algoritmo produce ridenominazioni sequenziali che possono influire negativamente sulle prestazioni. 1. Quando si scrive l'output in una partizione in una posizione personalizzata, le attività scrivono in un file nella directory di gestione temporanea di Spark, che viene creata nella posizione di output finale. Il nome del file include un UUID casuale per proteggere il file de collisioni. Il tentativo di attività tiene traccia di ogni file insieme al percorso di output finale desiderato. 1. Quando un'attività viene completata correttamente, fornisce al driver i file e i relativi percorsi di output finali desiderati. 1. Al termine di tutte le attività, la fase di commit dei lavori consente di rinominare in sequenza tutti i file scritti per le partizioni in percorsi personalizzati nei relativi percorsi di output finali. 1. La directory di gestione temporanea viene eliminata prima del completamento della fase di commit del processo. # Committer ottimizzato S3 EMRFS e caricamenti in più parti Per utilizzare il committer ottimizzato S3 EMRFS, i caricamenti in più parti devono essere abilitati in Amazon EMR. I caricamenti in più parti sono abilitati per impostazione predefinita. È possibile abilitarli nuovamente, se necessario. Per ulteriori informazioni, consulta [Configurazione del caricamento in più parti per Amazon S3](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart) nella *Guida alla gestione di Amazon EMR*. Il committer ottimizzato S3 EMRFS utilizza le caratteristiche tipo transazione dei caricamenti in più parti per garantire che i file scritti da tentativi di attività siano visualizzati solo nella posizione di output del processo durante il commit delle attività. Utilizzando caricamenti in più parti in questo modo, il committer migliora le prestazioni del task commit rispetto alla versione 2 dell'algoritmo predefinito. FileOutputCommitter Quando utilizzi il committer ottimizzato S3 EMRFS, ci sono delle differenze chiave rispetto al comportamento tradizionale del caricamento in più parti di cui tenere conto: + I caricamenti in più parti vengono sempre eseguiti indipendentemente dalle dimensioni del file. Questo differisce dal comportamento predefinito di EMRFS, dove la proprietà `fs.s3n.multipart.uploads.split.size` controlla le dimensioni del file in cui vengono attivati i caricamenti in più parti. + I caricamenti in più parti restano in stato incompleto per un periodo di tempo più lungo fino a quando avviene il commit o l'arresto dell'attività. Questo differisce dal comportamento predefinito di EMRFS dove un caricamento in più parti si competa quando l'attività termina la scrittura in un determinato file. A causa di queste differenze, se un Executor JVM Spark si arresta in modo anomalo o viene interrotto mentre le attività sono in corso e stanno scrivendo dati in Amazon S3, i caricamenti in più parti hanno più probabilità di restare indietro. Per questo motivo, quando utilizzi il committer ottimizzato S3 EMRFS, assicurati di seguire le best practice per la gestione dei caricamenti in più parti non riusciti. Per ulteriori informazioni, consulta le [best practice](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices) per utilizzare i bucket Amazon S3 nella *Guida alla gestione di Amazon EMR*. # Considerazioni sull'ottimizzazione dei processi Il committer ottimizzato S3 EMRFS consuma una piccola quantità di memoria per ogni file scritto da un tentativo di attività fino a quando avviene il commit o l'arresto dell'attività. Nella maggior parte dei processi la quantità di memoria consumata è trascurabile. Per i processi che presentano attività di lunga esecuzione che scrivono un numero elevato di file, la memoria che il committer consuma può essere notevole e richiede l'adeguamento della memoria allocata per gli executor Spark. Puoi ottimizzare la memoria dell'executor mediante la proprietà `spark.executor.memory`. Orientativamente, un'unica attività che scrive 100.000 file in genere richiede ulteriori 100 MB di memoria. Per ulteriori informazioni, consulta le [Proprietà delle applicazioni](https://spark.apache.org/docs/latest/configuration.html#application-properties) nella documentazione relativa alla configurazione di Apache Spark. # Abilitare il committer ottimizzato S3 EMRFS per Amazon EMR 5.19.0 Se usi Amazon EMR 5.19.0, puoi impostare manualmente la proprietà `spark.sql.parquet.fs.optimized.committer.optimization-enabled` su `true` quando crei un cluster o da Spark se stai utilizzando Amazon EMR. ## Abilitare il committer ottimizzato S3 EMRFS al momento della creazione di un cluster È possibile utilizzare la classificazione di configurazione `spark-defaults` per impostare la proprietà `spark.sql.parquet.fs.optimized.committer.optimization-enabled` su `true`. Per ulteriori informazioni, consulta [Configurazione delle applicazioni](emr-configure-apps.md). ## Abilitare il committer ottimizzato S3 EMRFS da Spark È possibile impostare `spark.sql.parquet.fs.optimized.committer.optimization-enabled` su `true` impostandola come hardcoded in un `SparkConf`, passandola come parametro `--conf` nella shell Spark o come strumenti `spark-submit` e `spark-sql`, o in `conf/spark-defaults.conf`. Per ulteriori informazioni, consulta la sezione relativa alla [Spark configuration (Configurazione Spark)](https://spark.apache.org/docs/latest/configuration.html) nella documentazione di Apache Spark. L'esempio seguente mostra come abilitare il committer durante l'esecuzione di un comando spark-sql. ``` spark-sql \ --conf spark.sql.parquet.fs.optimized.committer.optimization-enabled=true \ -e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;" ``` # Utilizzo del protocollo di commit ottimizzato per S3 EMRFS Il protocollo di commit ottimizzato per EMRFS S3 è un'[FileCommitProtocol](https://spark.apache.org/docs/2.2.0//api/java/org/apache/spark/internal/io/FileCommitProtocol.html)implementazione alternativa ottimizzata per scrivere file con sovrascrittura dinamica delle partizioni Spark su Amazon S3 quando si utilizza EMRFS. Il protocollo migliora le prestazioni delle applicazioni evitando le operazioni di ridenominazione in Amazon S3 durante la fase di commit del processo di sovrascrittura dinamica delle partizioni di Spark. Tieni presente che il committer ottimizzato per [EMRFS S3 migliora anche le prestazioni evitando le operazioni di ridenominazione](emr-spark-s3-optimized-committer.html). Tuttavia, non funziona per i casi di sovrascrittura dinamica delle partizioni, mentre i miglioramenti del protocollo commit riguardano solo i casi di sovrascrittura dinamica delle partizioni. Il protocollo di commit è disponibile in Amazon EMR rilascio 5.30.0 e successivi ed è abilitato per impostazione predefinita in Amazon EMR rilascio 6.2.0 e successivi. Amazon EMR ha aggiunto un miglioramento del parallelismo a partire dal rilascio 5.31.0. Il protocollo viene utilizzato per i lavori Spark che utilizzano Spark o Datasets. DataFrames Esistono circostanze in cui il protocollo di commit non viene utilizzato. Per ulteriori informazioni, consulta [Requisiti per il protocollo di commit ottimizzato per S3 EMRFS](emr-spark-committer-reqs.md). **Topics** + [Requisiti per il protocollo di commit ottimizzato per S3 EMRFS](emr-spark-commit-protocol-reqs.md) + [Protocollo di commit ottimizzato per S3 EMRFS e caricamenti in più parti](emr-spark-commit-protocol-multipart.md) + [Considerazioni sull'ottimizzazione dei processi](emr-spark-commit-protocol-tuning.md) # Requisiti per il protocollo di commit ottimizzato per S3 EMRFS Il protocollo di commit ottimizzato per S3 EMRFS viene utilizzato quando si verificano le condizioni riportate di seguito: + Si eseguono job Spark che utilizzano Spark o Datasets per sovrascrivere le DataFrames tabelle partizionate. + Vengono eseguiti processi Spark la cui modalità di sovrascrittura delle partizioni è `dynamic`. + I caricamenti in più parti sono abilitati in Amazon EMR. Questa è l’impostazione predefinita. Per ulteriori informazioni, consulta [Protocollo di commit ottimizzato per S3 EMRFS e caricamenti in più parti](emr-spark-commit-protocol-multipart.md). + La cache del file system per EMRFS è abilitata. Questa è l’impostazione predefinita. Verifica che l'impostazione `fs.s3.impl.disable.cache` sia impostata su `false`. + Viene utilizzato il supporto integrato per le origini dei dati di Spark. Il supporto integrato per le origini dei dati viene utilizzato nelle seguenti circostanze: + Quando i processi scrivono su origini dei dati o tabelle integrate. + Quando i processi scrivono tabelle Parquet del metastore Hive. Ciò accade quando `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastoreParquet` sono entrambi impostati su true. Queste sono le impostazioni predefinite. + Quando i processi scrivono tabelle ORC del metastore Hive. Ciò accade quando `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastoreOrc` sono entrambi impostati su `true`. Queste sono le impostazioni predefinite. + Le operazioni di processo Spark che scrivono in una posizione di partizione predefinita, ad esempio `${table_location}/k1=v1/k2=v2/`, usano il protocollo di commit. Il protocollo non viene utilizzato se un'operazione di processo scrive in una posizione di partizione personalizzata, ad esempio se un percorso di partizione personalizzato è impostato utilizzando il comando `ALTER TABLE SQL`. + Devono essere utilizzati i seguenti valori per Spark: + `spark.sql.sources.commitProtocolClass` deve essere impostato su `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol`. Questa è l'impostazione predefinita per Amazon EMR rilascio 5.30.0 e successivi, e rilascio 6.2.0 e successivi. + L'opzione di scrittura `partitionOverwriteMode` o `spark.sql.sources.partitionOverwriteMode` deve essere impostata su `dynamic`. L'impostazione predefinita è `static`. **Nota** L'opzione di scrittura `partitionOverwriteMode` è stata introdotta in Spark 2.4.0. Per Spark versione 2.3.2, incluso con Amazon EMR rilascio 5.19.0, imposta la proprietà `spark.sql.sources.partitionOverwriteMode`. + Se i processi Spark sovrascrivono la tabella Parquet del metastore Hive, `spark.sql.hive.convertMetastoreParquet`, `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastore.partitionOverwriteMode` devono essere impostati su `true`. Queste sono le impostazioni predefinite. + Se i processi Spark sovrascrivono la tabella ORC del metastore Hive, `spark.sql.hive.convertMetastoreOrc`, `spark.sql.hive.convertInsertingPartitionedTable` e `spark.sql.hive.convertMetastore.partitionOverwriteMode` devono essere impostati su `true`. Queste sono le impostazioni predefinite. **Example - Modalità di sovrascrittura dinamica delle partizioni** In questo esempio di Scala, viene attivata l'ottimizzazione. Innanzitutto, imposta la proprietà `partitionOverwriteMode` su `dynamic`. Questo sovrascrive solo le partizioni in cui stai scrivendo i dati. Quindi, specifichi le colonne delle partizioni dinamiche con `partitionBy` e imposti la modalità scrittura su `overwrite`. ``` val dataset = spark.range(0, 10) .withColumn("dt", expr("date_sub(current_date(), id)")) dataset.write.mode("overwrite") // "overwrite" instead of "insert" .option("partitionOverwriteMode", "dynamic") // "dynamic" instead of "static" .partitionBy("dt") // partitioned data instead of unpartitioned data .parquet("s3://amzn-s3-demo-bucket1/output") // "s3://" to use Amazon EMR file system, instead of "s3a://" or "hdfs://" ``` ## Quando il protocollo di commit ottimizzato per S3 EMRFS non viene utilizzato In genere, il protocollo di commit ottimizzato per EMRFS S3 funziona allo stesso modo del protocollo di commit Spark predefinito open source,. `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` L'ottimizzazione non si verifica nelle seguenti situazioni. **** | Situazione | Perché il protocollo di commit non viene utilizzato | | --- | --- | | Quando si scrive su HDFS | Il protocollo di commit supporta solo la scrittura su Amazon S3 utilizzando EMRFS. | | Quando si utilizza il file system S3A | Il protocollo di commit supporta solo EMRFS. | | Quando utilizzi MapReduce la nostra API RDD di Spark | Il protocollo di commit supporta solo l'utilizzo di SparkSQL o DataFrame Dataset. APIs | | Quando la sovrascrittura dinamica delle partizioni non viene attivata | Il protocollo di commit ottimizza solo i casi di sovrascrittura dinamica delle partizioni. Per altri casi, consulta la sezione [Utilizzare il committer ottimizzato S3 EMRFS](emr-spark-s3-optimized-committer.md). | I seguenti esempi di Scala mostrano altre situazioni che il protocollo di commit ottimizzato per S3 EMRFS delega `SQLHadoopMapReduceCommitProtocol`. **Example - Modalità di sovrascrittura delle partizioni con posizione della partizione personalizzata** In questo esempio, i programmi Scala sovrascrivono due partizioni in modalità di sovrascrittura dinamica delle partizioni. Una partizione ha una posizione di partizione personalizzata. L'altra partizione usa il percorso di partizione predefinito. Il protocollo di commit ottimizzato per S3 EMRFS viene utilizzato solo per scrivere l'output dell'attività nella partizione che utilizza la posizione della partizione predefinita. ``` val table = "dataset" val inputView = "tempView" val location = "s3://bucket/table" spark.sql(s""" CREATE TABLE $table (id bigint, dt date) USING PARQUET PARTITIONED BY (dt) LOCATION '$location' """) // Add a partition using a custom location val customPartitionLocation = "s3://bucket/custom" spark.sql(s""" ALTER TABLE $table ADD PARTITION (dt='2019-01-28') LOCATION '$customPartitionLocation' """) // Add another partition using default location spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')") def asDate(text: String) = lit(text).cast("date") spark.range(0, 10) .withColumn("dt", when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29"))) .createTempView(inputView) // Set partition overwrite mode to 'dynamic' spark.sql(s"SET spark.sql.sources.partitionOverwriteMode=dynamic") spark.sql(s"INSERT OVERWRITE TABLE $table SELECT * FROM $inputView") ``` Il codice Scala crea i seguenti oggetti Amazon S3: ``` custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet custom_$folder$ table/_SUCCESS table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet table/dt=2019-01-29_$folder$ table_$folder$ ``` La scrittura in posizioni di partizione personalizzate nelle versioni precedenti di Spark può causare la perdita di dati. In questo esempio, la partizione `dt='2019-01-28'` andrebbe persa. Per ulteriori dettagli, consulta [SPARK-35106](https://issues.apache.org/jira/browse/SPARK-35106). Questo problema è stato risolto in Amazon EMR rilascio 5.33.0 e successivi, esclusi i rilasci 6.0.x e 6.1.x. Quando si scrive nelle partizioni in posizioni personalizzate, Spark utilizza un algoritmo di commit simile all'esempio precedente, che è descritto di seguito. Come nell'esempio precedente, l'algoritmo produce ridenominazioni sequenziali che possono influire negativamente sulle prestazioni. L'algoritmo in Spark 2.4.0 segue questi passaggi: 1. Quando si scrive l'output in una partizione in una posizione personalizzata, le attività scrivono in un file nella directory di gestione temporanea di Spark, che viene creata nella posizione di output finale. Il nome del file include un UUID casuale per proteggere il file de collisioni. Il tentativo di attività tiene traccia di ogni file insieme al percorso di output finale desiderato. 1. Quando un'attività viene completata correttamente, fornisce al driver i file e i relativi percorsi di output finali desiderati. 1. Al termine di tutte le attività, la fase di commit dei lavori consente di rinominare in sequenza tutti i file scritti per le partizioni in percorsi personalizzati nei relativi percorsi di output finali. 1. La directory di gestione temporanea viene eliminata prima del completamento della fase di commit del processo. # Protocollo di commit ottimizzato per S3 EMRFS e caricamenti in più parti Per utilizzare l'ottimizzazione per la sovrascrittura dinamica delle partizioni nel protocollo di commit ottimizzato per S3 EMRFS, i caricamenti in più parti devono essere abilitati in Amazon EMR. I caricamenti in più parti sono abilitati per impostazione predefinita. È possibile abilitarli nuovamente, se necessario. Per ulteriori informazioni, consulta [Configurazione del caricamento in più parti per Amazon S3](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart) nella *Guida alla gestione di Amazon EMR*. Durante la scovrascittura dinamica delle partizioni, il protocollo di commit ottimizzato per S3 EMRFS utilizza le caratteristiche tipo transazione dei caricamenti in più parti per garantire che i file scritti da tentativi di attività siano visualizzati solo nella posizione di output del processo durante il commit del processo. Utilizzando i caricamenti in più parti in questo modo, il protocollo di commit migliora le prestazioni del commit dei processi rispetto al protocollo `SQLHadoopMapReduceCommitProtocol` predefinito. Quando utilizzi il protocollo di commit ottimizzato per S3 EMRFS, esistono delle differenze chiave rispetto al comportamento tradizionale del caricamento in più parti di cui tenere conto: + I caricamenti in più parti vengono sempre eseguiti indipendentemente dalle dimensioni del file. Questo differisce dal comportamento predefinito di EMRFS, dove la proprietà `fs.s3n.multipart.uploads.split.size` controlla le dimensioni del file in cui vengono attivati i caricamenti in più parti. + I caricamenti in più parti restano in stato incompleto per un periodo di tempo più lungo fino a quando avviene il commit o l'arresto dell'attività. Questo differisce dal comportamento predefinito di EMRFS dove un caricamento in più parti si competa quando l'attività termina la scrittura in un determinato file. A causa di queste differenze, se un esecutore JVM di Spark si arresta in modo anomalo o viene interrotto mentre le attività sono in corso e stanno scrivendo dati in Amazon S3, oppure un driver JVM di Spark ha un arresto anomalo o viene chiuso in modo forzato mentre è in corso un processo, i caricamenti in più parti hanno più probabilità di restare indietro. Per questo motivo, quando utilizzi il protocollo di commit ottimizzato per S3 EMRFS, assicurati di seguire le best practice per la gestione dei caricamenti in più parti non riusciti. Per ulteriori informazioni, consulta le [best practice](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices) per utilizzare i bucket Amazon S3 nella *Guida alla gestione di Amazon EMR*. # Considerazioni sull'ottimizzazione dei processi Sugli esecutori Spark, il protocollo di commit ottimizzato per S3 EMRFS consuma una piccola quantità di memoria per ogni file scritto da un tentativo di attività fino a quando avviene il commit o l'arresto dell'attività. Nella maggior parte dei processi la quantità di memoria consumata è trascurabile. Sui driver Spark, il protocollo di commit ottimizzato per S3 EMRFS richiede memoria per archiviare le informazioni sui metadati di ogni file salvato fino a quando avviene il commit o l'arresto del processo. Nella maggior parte dei processi, l'impostazione predefinita della memoria del driver Spark è trascurabile. Per i processi che presentano attività di lunga esecuzione che scrivono un numero elevato di file, la memoria che il protocollo di commit consuma può essere notevole e rende necessario adeguare la memoria allocata per Spark, specialmente per gli esecutori Spark. È possibile regolare la memoria utilizzando la proprietà `spark.driver.memory` per i driver Spark e la proprietà `spark.executor.memory` per gli esecutori Spark. Orientativamente, un'unica attività che scrive 100.000 file in genere richiede ulteriori 100 MB di memoria. Per ulteriori informazioni, consulta le [Proprietà delle applicazioni](https://spark.apache.org/docs/latest/configuration.html#application-properties) nella documentazione relativa alla configurazione di Apache Spark. # Effettuare nuovi tentativi per le richieste Amazon S3 con EMRFS In questo argomento vengono fornite informazioni sulle strategie di nuovi tentativi che è possibile utilizzare quando si effettuano richieste ad Amazon S3 con EMRFS. Quando il tasso di richiesta aumenta, S3 tenta di dimensionarsi per supportare la nuova tariffa. Durante questo processo, S3 può limitare le richieste e restituire un errore `503 Slow Down`. Per migliorare il tasso di successo delle richieste S3, è possibile modificare la strategia di nuovi tentativi impostando le proprietà nella configurazione di `emrfs-site` È possibile modificare la strategia di nuovi tentativi nei modi seguenti. + Aumentare il limite massimo di tentativi per la strategia di nuovi tentativi con backoff esponenziale predefinita. + Abilitare e configurare la strategia di nuovi tentativi come aumento additivo - diminuzione moltiplicativa (AIMD). AIMD è supportato per Amazon EMR rilascio 6.4.0 e successivi. ## Usare la strategia di backoff esponenziale predefinita Per impostazione predefinita, EMRFS utilizza una strategia di backoff esponenziale per effettuare nuovi tentativi per le richieste Amazon S3. Il limite di nuovi tentativi predefinito di EMRFS è pari a 15. Per evitare un errore `503 Slow Down` di S3 è possibile aumentare il limite di nuovi tentativi quando si crea un nuovo cluster, in un cluster in esecuzione o in fase di runtime dell'applicazione. Per aumentare il limite di nuovi tentativi è necessario modificare il valore di `fs.s3.maxRetries` nella propria configurazione `emrfs-site`. I seguenti set di configurazione di esempio impostano `fs.s3.maxRetries` a un valore personalizzato pari a 30. ``` [ { "Classification": "emrfs-site", "Properties": { "fs.s3.maxRetries": "30" } } ] ``` Per ulteriori informazioni sull'utilizzo degli oggetti di configurazione, consulta [Configurazione delle applicazioni](emr-configure-apps.md). ## Usare la strategia di nuovi tentativi dell'AIMD Con Amazon EMR rilascio 6.4.0 e successivi, EMRFS supporta una strategia alternativa di ripetizione basata su un modello di aumento additivo/diminuzione moltiplicativa (AIMD). La strategia di nuovi tentativi AIMD è particolarmente utile quando si lavora con cluster Amazon EMR di grandi dimensioni. AIMD calcola una frequenza di richiesta personalizzata utilizzando i dati sulle recenti richieste di successo. Questa strategia riduce il numero di richieste limitate e i tentativi totali richiesti per ciascuna richiesta. Per abilitare la strategia di nuovi tentativi AIMD, è necessario impostare la proprietà `fs.s3.aimd.enabled` su `true` nella configurazione `emrfs-site` come nell'esempio seguente. ``` [ { "Classification": "emrfs-site", "Properties": { "fs.s3.aimd.enabled": "true" } } ] ``` Per ulteriori informazioni sull'utilizzo degli oggetti di configurazione, consulta [Configurazione delle applicazioni](emr-configure-apps.md). ## Impostazioni avanzate di nuovi tentativi dell'AIMD È possibile configurare le proprietà elencate nella tabella seguente per perfezionare il comportamento dei nuovi tentativi quando si utilizza la strategia di nuovi tentativi dell'AIMD. Per la maggior parte dei casi d'uso, si consiglia di utilizzare i valori predefiniti. **Proprietà avanzate della strategia di nuovi tentativi dell'AIMD** | Proprietà | Valore predefinito | Description | | --- | --- | --- | | fs.s3.aimd.increaseIncrement | 0.1 | Controlla la rapidità con cui il tasso di richiesta aumenta quando le richieste consecutive hanno esito positivo. | | fs.s3.aimd.reductionFactor | 2 | Controlla la rapidità con cui il tasso di richiesta diminuisce quando Amazon S3 restituisce una risposta 503. Il fattore predefinito di 2 riduce a metà il tasso di richiesta. | | fs.s3.aimd.minRate | 0.1 | Imposta il limite inferiore per il tasso di richiesta quando le richieste subiscono una limitazione sostenuta da S3. | | fs.s3.aimd.initialRate | 5500 | Imposta il tasso di richiesta iniziale, che cambia in base ai valori specificati per fs.s3.aimd.increaseIncrement e fs.s3.aimd.reductionFactor.La frequenza iniziale viene utilizzata anche per le richieste GET e ridimensionata proporzionalmente (3500/5500) per le richieste PUT. | | fs.s3.aimd.adjustWindow | 2 | Controlla la frequenza di adeguamento del tasso di richiesta, misurato in numero di risposte. | | fs.s3.aimd.maxAttempts | 100 | Imposta il numero massimo di tentativi di provare una richiesta. |