Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Utilisation d'un valideur EMRFS optimisé pour S3
Le committer optimisé pour EMRFS S3 est une [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html)implémentation alternative optimisée pour écrire des fichiers sur Amazon S3 lors de l'utilisation d'EMRFS. Le validateur EMRFS optimisé pour S3 améliore les performances de l'application en évitant les opérations de liste et de renommage effectuées dans Amazon S3 pendant les phases de validation des tâches. Le valideur est disponible dans les versions 5.19.0 et ultérieures d'Amazon EMR et est activé par défaut dans les versions 5.20.0 et ultérieures d'Amazon EMR. Le committer est utilisé pour les tâches Spark qui utilisent Spark ou Datasets. DataFrames À partir d'Amazon EMR 6.4.0, ce validateur peut être utilisé pour tous les formats courants, notamment parquet, ORC et les formats basés sur le texte (notamment CSV et JSON). Pour les versions antérieures à Amazon EMR 6.4.0, seul le format Parquet est pris en charge. Dans certaines situations, le valideur n'est pas utilisé. Pour de plus amples informations, veuillez consulter [Exigences du valideur EMRFS optimisé pour S3](emr-spark-committer-reqs.md).
**Topics**
+ [Exigences du valideur EMRFS optimisé pour S3](emr-spark-committer-reqs.md)
+ [Le valideur EMRFS optimisé pour S3 et les chargements partitionnés](emr-spark-committer-multipart.md)
+ [Considérations de réglage de tâche](emr-spark-committer-tuning.md)
+ [Activer le validateur EMRFS S3 pour Amazon EMR 5.19.0](emr-spark-committer-enable.md)
# Exigences du valideur EMRFS optimisé pour S3
Le valideur EMRFS optimisé pour S3 est utilisé lorsque les conditions suivantes sont réunies :
+ Vous exécutez des tâches Spark qui utilisent Spark ou Datasets pour écrire des fichiers sur Amazon S3. DataFrames À partir d'Amazon EMR 6.4.0, ce validateur peut être utilisé pour tous les formats courants, notamment parquet, ORC et les formats basés sur le texte (notamment CSV et JSON). Pour les versions antérieures à Amazon EMR 6.4.0, seul le format Parquet est pris en charge.
+ Les chargements partitionnés sont activés dans Amazon EMR. Il s’agit de l’option par défaut. Pour de plus amples informations, veuillez consulter [Le valideur EMRFS optimisé pour S3 et les chargements partitionnés](emr-spark-committer-multipart.md).
+ Le support de format de fichier intégré à Spark est utilisé. La prise en charge du format de fichier intégré est utilisée dans les circonstances suivantes :
+ Pour les tables de métastore Hive, lorsque `spark.sql.hive.convertMetastoreParquet` est défini sur `true` pour les tables Parquet, ou lorsque `spark.sql.hive.convertMetastoreOrc` est défini sur `true` pour les tables Orc avec Amazon EMR 6.4.0 ou version ultérieure. Il s'agit des paramètres par défaut.
+ Lorsque les tâches écrivent dans des sources de données ou des tables au format de fichier – par exemple, la table cible est créée avec la clause `USING parquet`.
+ Lorsque des tâches écrivent dans des tableaux Parquet de metastore Hive non partitionnées. Le support Parquet intégré à Spark ne prend pas en charge avec les tableaux Hive partitionnés. Il s'agit d'une limitation connue. Pour plus d'informations, consultez la section [Conversion des tables Hive Metastore Parquet](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#hive-metastore-parquet-table-conversion) dans le guide Apache Spark DataFrames et Datasets.
+ Les opérations de tâche Spark qui écrivent dans un emplacement de partition par défaut (par exemple, `${table_location}/k1=v1/k2=v2/`) utilisent le valideur. Le valideur n'est pas utilisé si une opération de tâche écrit dans un emplacement de partition personnalisé, par exemple, si un emplacement de partition personnalisé est défini à l'aide de la commande `ALTER TABLE SQL`.
+ Les valeurs suivantes pour Spark doivent être utilisées :
+ La propriété `spark.sql.parquet.fs.optimized.committer.optimization-enabled` doit être définie sur `true`. Il s'agit du paramètre par défaut avec Amazon EMR 5.20.0 ou version suivante. Avec Amazon EMR 5.19.0, la valeur par défaut est `false`. Pour plus d'informations sur la configuration de cette valeur, consultez [Activer le validateur EMRFS S3 pour Amazon EMR 5.19.0](emr-spark-committer-enable.md).
+ Si vous écrivez dans des tables de métastore Hive non partitionnées, seuls les formats de fichier Parquet et Orc sont pris en charge. `spark.sql.hive.convertMetastoreParquet`doit être défini sur `true` si vous écrivez sur des tables de métastore Parquet Hive non partitionnées. `spark.sql.hive.convertMetastoreOrc`doit être défini sur `true` si vous écrivez dans des tables de métastore Orc Hive non partitionnées. Il s'agit des paramètres par défaut.
+ `spark.sql.parquet.output.committer.class` doit être défini sur `com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter`. Il s’agit du paramètre par défaut.
+ `spark.sql.sources.commitProtocolClass` doit être défini sur `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` ou `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol`. `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` est le paramètre par défaut pour les versions 5.30.0 et ultérieures de la série Amazon EMR 5.x, et pour les versions 6.2.0 et ultérieures de la série Amazon EMR 6.x. `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` est le paramètre par défaut pour les versions précédentes d’Amazon EMR.
+ Si des tâches Spark remplacent des ensembles de données Parquet partitionnés par des colonnes de partition dynamique, l'option d'écriture `partitionOverwriteMode` et `spark.sql.sources.partitionOverwriteMode` doivent être définis sur `static`. Il s’agit du paramètre par défaut.
**Note**
L'option d'écriture `partitionOverwriteMode` a été introduite dans Spark 2.4.0. Pour Spark version 2.3.2, inclus avec Amazon EMR version 5.19.0, définissez la propriété `spark.sql.sources.partitionOverwriteMode`.
## Cas dans lesquels le valideur EMRFS optimisé pour S3 n’est pas utilisé
En général, le valideur EMRFS S3 optimisé n’est pas utilisé dans les situations suivantes.
****
| Situation | Pourquoi le validateur n'est pas utilisé |
| --- | --- |
| Lorsque vous écrivez dans HDFS | Le validateur prend uniquement en charge l'écriture sur Amazon S3 à l'aide d'EMRFS. |
| Lorsque vous utilisez le système de fichiers S3A | Le validateur ne prend en charge que EMRFS. |
| Lorsque vous utilisez l' MapReduce API RDD de Spark | Le committer prend uniquement en charge l'utilisation de SparkSQL ou DataFrame Dataset. APIs |
Les exemples en Scala suivants illustrent quelques situations supplémentaires qui empêchent d'utiliser le validateur optimisé pour EMRFS S3 en totalité (premier exemple) et en partie (deuxième exemple).
**Example – Mode de remplacement de partition dynamique**
L'exemple Scala suivant indique à Spark d'utiliser un algorithme de validation différent, ce qui empêche complètement l'utilisation du validateur optimisé pour EMRFS S3. Le code définit la propriété `partitionOverwriteMode` sur `dynamic` pour n'écraser que les partitions sur lesquelles vous écrivez des données. Ensuite, les colonnes de partition dynamique sont spécifiées par `partitionBy`, et le mode d'écriture est défini sur `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")
```
Vous devez configurer les trois paramètres pour éviter d'utiliser le validateur EMRFS S3. Lorsque vous le faites, Spark exécute un algorithme de validation différent qui est spécifié dans le protocole de validation de Spark. Pour les versions d'Amazon EMR 5.x antérieures à la version 5.30.0 et pour les versions d'Amazon EMR 6.x antérieures à la version 6.2.0, le protocole de validation utilise le répertoire de préparation de Spark, qui est un répertoire temporaire créé sous l'emplacement de sortie commençant par `.spark-staging`. L'algorithme renomme séquentiellement les répertoires de partition, ce qui peut avoir un impact négatif sur les performances. Pour de plus amples informations sur Amazon EMR versions 5.30.0 et versions ultérieures, ainsi que 6.2.0 et versions ultérieures, veuillez consulter [Utiliser le protocole de validation optimisé pour EMRFS S3](emr-spark-s3-optimized-commit-protocol.md).
L'algorithme dans Spark 2.4.0 exécute les étapes suivantes :
1. Les tentatives de tâches écrivent leur résultat dans les répertoires de partition situés dans le répertoire de préparation de Spark, par exemple, `${outputLocation}/spark-staging-${jobID}/k1=v1/k2=v2/`.
1. Pour chaque partition écrite, la tâche tente de suivre les chemins de partition relatifs, par exemple, `k1=v1/k2=v2`.
1. Lorsqu'une tâche se termine avec succès elle fournit le pilote avec tous les chemins de partition relatifs qu'elle a suivis.
1. Une fois toutes les tâches terminées, la phase de validation de tâche collecte tous les répertoires de partition que les tentatives de tâche réussies ont écrit dans le répertoire intermédiaire de Spark. Spark renomme de manière séquentielle chacun de ces répertoires dans son emplacement de sortie final à l'aide d'opérations attribution de nouveau nom de l'arborescence.
1. Le répertoire intermédiaire est supprimé avant que la phase de validation de tâche soit terminée.
**Example – Emplacement de partition personnalisé**
Dans cet exemple, le code Scala s'insère dans deux partitions. Une partition possède un emplacement de partition personnalisé. L'autre partition utilise l'emplacement de partition par défaut. Le valideur EMRFS optimisé pour S3 est utilisé uniquement pour écrire la sortie de tâche dans la partition qui utilise l'emplacement de partition par défaut.
```
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)
```
Le code Scala crée les objets Amazon S3 suivants :
```
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$
```
Lorsque vous écrivez des partitions dans des emplacements personnalisés, Spark utilise un algorithme de validation similaire à celui de l'exemple précédent, qui est décrit ci-dessous. Comme dans l'exemple précédent, l'algorithme se traduit par des attributions séquentielles de nouveaux noms, ce qui peut avoir un impact négatif sur les performances.
1. Lors de l'écriture de la sortie d'une partition dans un emplacement personnalisé, les tâches écrivent un fichier dans le répertoire intermédiaire de Spark, qui est créé sous l'emplacement de sortie final. Le nom du fichier comprend un UUID aléatoire pour éviter les conflits de fichier. La tentative de tâche suit chaque fichier, ainsi que le chemin de la sortie souhaité final.
1. Lorsqu'une tâche se termine avec succès, elle fournit au pilote les fichiers et les chemins de sortie souhaités finaux.
1. Une fois toutes les tâches terminées, la phase de validation de tâche renomme de manière séquentielle tous les fichiers qui ont été écrits pour les partitions dans les emplacements personnalisés en leurs chemins de sortie finaux.
1. Le répertoire intermédiaire est supprimé avant que la phase de validation de tâche soit terminée.
# Le valideur EMRFS optimisé pour S3 et les chargements partitionnés
Pour utiliser le valideur EMRFS optimisé pour S3, vous devez activer les chargements partitionnés dans Amazon EMR. Les chargements partitionnés sont activés par défaut. Vous pouvez les réactiver si besoin est. Pour plus d'informations, consultez [Configuration d'un chargement partitionné pour Simple Storage Service (Amazon S3)](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart) dans le *Guide de gestion Amazon EMR*.
Le valideur EMRFS optimisé pour S3 utilise des caractéristiques similaires à des transactions de chargements partitionnés pour assurer que les fichiers écrits par des tentatives de tâches apparaissent uniquement dans l'emplacement de sortie du travail lors de la validation de tâche. En utilisant les téléchargements partitionnés de cette manière, le validateur améliore les performances de validation des tâches par rapport à la version 2 de l' FileOutputCommitter algorithme par défaut. Lors de l'utilisation d'un valideur EMRFS optimisé pour S3, il existe certaines différences importantes à prendre en compte comparé au comportement de chargement partitionné :
+ Les chargements partitionnés sont toujours effectués, peu importe la taille du fichier. Ceci diffère du comportement normal d'EMRFS, où la propriété `fs.s3n.multipart.uploads.split.size` contrôle la taille du fichier à laquelle les chargements partitionnés sont enclenchés.
+ Les chargements partitionnés sont laissés en état non terminé pendant un plus long laps de temps jusqu'à ce que la tâche soit validée ou abandonnée. Ceci diffère du comportement normal d'EMRFS, où un chargement partitionné se termine lorsqu'un tâche termine d'écrire un fichier donné.
En raison de ces différences, si un Spark Executor JVM crache ou qu'il est tué lorsque des tâches sont en cours d'exécution ou écrivent des données à Amazon S3, les chargements partitionnés inachevés ont plus de chances d'être laissés de côté. Pour cette raison, lorsque vous utilisez un valideur EMRFS optimisé pour S3, assurez-vous de suivre les bonnes pratiques de gestion de chargements partitionnés échoués. Pour plus d'informations, consultez [Bonnes pratiques](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices) pour travailler avec des compartiments Amazon S3 dans le *Guide de gestion Amazon EMR*.
# Considérations de réglage de tâche
Le valideur EMRFS optimisé pour S3 consomme une petite quantité de mémoire pour chaque fichier écrit par une tentative de tâche, jusqu'à ce que la tâche soit validée ou abandonnée. La quantité de mémoire consommée est négligeable dans la plupart des tâches. Pour les tâches dotées de tâches à longue exécution qui écrivent un grand nombre de fichiers, la mémoire consommée par le valideur peut être significative et nécessitera des ajustements à l'allocation de mémoire aux exécuteurs Spark. Vous pouvez ajuster la mémoire d'exécuteur à l'aide de la propriété `spark.executor.memory`. Une seule tâche écrivant 100 000 fichiers nécessiterait en général environ 100 Mo de mémoire supplémentaire. Pour plus d'informations, consultez [Propriétés d'applications](https://spark.apache.org/docs/latest/configuration.html#application-properties) dans la Documentation de configuration Apache Spark.
# Activer le validateur EMRFS S3 pour Amazon EMR 5.19.0
Si vous utilisez Amazon EMR 5.19.0, vous pouvez définir la propriété `spark.sql.parquet.fs.optimized.committer.optimization-enabled` sur `true` lorsque vous créez un cluster, ou depuis Spark si vous utilisez Amazon EMR.
## Activation d'un valideur EMRFS optimisé pour S3 lors de la création d'un cluster
Utilisez la classification de configuration `spark-defaults` pour définir la propriété `spark.sql.parquet.fs.optimized.committer.optimization-enabled` sur `true`. Pour de plus amples informations, veuillez consulter [Configuration des applications](emr-configure-apps.md).
## Activation d'un valideur EMRFS optimisé pour S3 depuis Spark
Vous pouvez définir `spark.sql.parquet.fs.optimized.committer.optimization-enabled` sur `true` en l'ancrant dans un `SparkConf`, en le passant comme paramètre `--conf` dans le shell Spark ou les outils `spark-submit` et `spark-sql`, ou dans `conf/spark-defaults.conf`. Pour plus d'informations, consultez [Configuration Spark](https://spark.apache.org/docs/latest/configuration.html) dans la documentation Apache Spark.
L'exemple suivant montre comment activer le valideur en exécutant une commande spark-sql.
```
spark-sql \
--conf spark.sql.parquet.fs.optimized.committer.optimization-enabled=true \
-e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;"
```