Quero criar um pipeline de processamento de dados na AWS para, eventualmente, usar os dados processados para o Machine Learning.

Eu tenho um script Scala que pega dados brutos do S3, os processa e os grava no HDFS ou mesmo no S3 comSpark-CSV. Eu acho que posso usar vários arquivos como entrada, se eu quiser usarAprendizado de máquina da AWS ferramenta para treinar um modelo de previsão. Mas se eu quiser usar outra coisa, presumo que seja melhor receber um único arquivo de saída CSV.

Atualmente, como eu não quero usarreparticionamento (1) nemcoalescer (1) para fins de desempenho, eu useihadoop fs -getmerge para teste manual, mas como ele mescla apenas o conteúdo dos arquivos de saída do trabalho, estou com um pequeno problema. eu precisouma única linha de cabeçalhos no arquivo de dados para treinar o modelo de previsão.

Se eu usar.option("header","true") para o spark-csv, ele grava os cabeçalhos em todos os arquivos de saída e, depois da mesclagem, tenho tantas linhas de cabeçalhos nos dados quanto os arquivos de saída. Mas se a opção do cabeçalho for falsa, ela não adicionará nenhum cabeçalho.

Agora eu encontrei uma opção para mesclar os arquivos dentro do script Scala com a API do HadoopFileUtil.copyMerge. Eu tentei isso emspark-shell com o código abaixo.

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, "")

Mas essa solução ainda concatena os arquivos uns sobre os outros e não manipula cabeçalhos.Como posso obter um arquivo de saída com apenas uma linha de cabeçalhos?

Eu até tentei adicionardf.columns.mkString(",") como o último argumento paracopyMerge, mas isso adicionou os cabeçalhos ainda várias vezes, nem uma vez.