Lo que está viendo es un problema con outputcommitter y s3. el trabajo de confirmación aplica fs.rename en la carpeta _temporal y dado que S3 no admite el cambio de nombre, significa que una sola solicitud ahora está copiando y eliminando todos los archivos de _temporal a su destino final.

sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.algorithm.version", "2") solo funciona con la versión de hadoop > 2.7. lo que hace es copiar cada archivo de _temporal en la tarea de confirmación y no confirmar la tarea para que se distribuya y funcione bastante rápido.

Si usa una versión anterior de Hadoop, usaría Spark 1.6 y usaría:

 sc.hadoopConfiguration.set("spark.sql.parquet.output.committer.class","org.apache.spark.sql.parquet.DirectParquetOutputCommitter")

* tenga en cuenta que no funciona con la especulación activada o escribiendo en modo de adición

**También tenga en cuenta que está en desuso en Spark 2.0 (reemplazado por algoritmo.versión=2)

Por cierto, en mi equipo escribimos con Spark a HDFS y usamos trabajos DISTCP (específicamente s3-dist-cp) en producción para copiar los archivos a S3, pero esto se hace por varias otras razones (coherencia, tolerancia a fallas), por lo que no es necesario .. puedes escribir en S3 bastante rápido usando lo que sugerí.

Tuve un caso de uso similar en el que usé chispa para escribir en s3 y tuve un problema de rendimiento. La razón principal fue que Spark estaba creando muchos archivos parciales de cero bytes y reemplazar los archivos temporales por el nombre real del archivo estaba ralentizando el proceso de escritura. Intenté el siguiente enfoque como solución alternativa

1. Escriba la salida de Spark en HDFS y use Hive para escribir en s3. El rendimiento fue mucho mejor ya que Hive creaba menos archivos de piezas. El problema que tuve es (también tuve el mismo problema al usar Spark), la acción de eliminación en la política no se proporcionó en prod env por razones de seguridad. En mi caso, el cubo S3 estaba encriptado en kms.

2. Escriba la salida de chispa en HDFS y copie los archivos hdfs en local y use aws s3 copy para enviar datos a s3. Tuvo los segundos mejores resultados con este enfoque. Boleto creado con Amazon y sugirieron ir con este.

3. Use s3 dist cp para copiar archivos de HDFS a S3. Esto estaba funcionando sin problemas, pero sin rendimiento

El comunicador directo fue sacado de la chispa ya que no era resistente a las fallas. Recomiendo encarecidamente no usarlo.

Se está trabajando en Hadoop, s3guard, para agregar 0-rename committers, que serán O(1) y tolerantes a fallas; mantener un ojo en HADOOP-13786 .

Ignorando "el confirmador de Magic" por ahora, el confirmador de pruebas basado en Netflix se enviará primero (¿hadoop 2.9? ¿3.0?)

1. Esto escribe el trabajo en el FS local, en la tarea de confirmación.
2. emite operaciones put de varias partes no comprometidas para escribir los datos, pero no materializarlos.
3. guarda la información necesaria para enviar el PUT a HDFS, utilizando el confirmador de salida de archivo "algoritmo 1" original
4. Implementa una confirmación de trabajo que utiliza la confirmación de salida del archivo de HDFS para decidir qué PUT completar y cuáles cancelar.

Resultado: la confirmación de la tarea toma segundos de datos/ancho de banda, pero la confirmación del trabajo no toma más tiempo que hacer 1-4 GET en la carpeta de destino y un POST para cada archivo pendiente, este último en paralelo.

Puede elegir el committer en el que se basa este trabajo, de netflix , y probablemente usarlo en Spark hoy. Establezca el algoritmo de confirmación del archivo = 1 (debería ser el valor predeterminado) o en realidad no escribirá los datos.