Quiero crear una tubería de procesamiento de datos en AWS para eventualmente usar los datos procesados para Machine Learning.

Tengo un script Scala que toma datos sin procesar de S3, los procesa y los escribe en HDFS o incluso S3 conSpark-CSV. Creo que puedo usar varios archivos como entrada si quiero usarAprendizaje automático de AWS herramienta para entrenar un modelo de predicción. Pero si quiero usar algo más, supongo que es mejor si recibo un solo archivo de salida CSV.

Actualmente, como no quiero usarrepartición (1) nifusionarse (1) para fines de rendimiento, he usadohadoop fs -getmerge para pruebas manuales, pero como solo combina el contenido de los archivos de salida del trabajo, me encuentro con un pequeño problema. Necesitouna sola fila de encabezados en el archivo de datos para entrenar el modelo de predicción.

Si yo uso.option("header","true") para el spark-csv, luego escribe los encabezados en cada archivo de salida y después de fusionar tengo tantas líneas de encabezados en los datos como archivos de salida. Pero si la opción de encabezado es falsa, entonces no agrega ningún encabezado.

Ahora encontré una opción para fusionar los archivos dentro del script Scala con la API de HadoopFileUtil.copyMerge. Intenté esto enspark-shell con el siguiente código.

import org.apache.hadoop.fs.FileUtil
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
val configuration = new Configuration();
val fs = FileSystem.get(configuration);
FileUtil.copyMerge(fs, new Path("smallheaders"), fs, new Path("/home/hadoop/smallheaders2"), false, configuration, "")

Pero esta solución aún concatena los archivos uno encima del otro y no maneja encabezados.¿Cómo puedo obtener un archivo de salida con solo una fila de encabezados?

Incluso intenté agregardf.columns.mkString(",") como el último argumento paracopyMerge, pero esto agregó los encabezados aún varias veces, no una.