Ich möchte eine Datenverarbeitungs-Pipeline in AWS erstellen, um die verarbeiteten Daten schließlich für maschinelles Lernen zu verwenden.

Ich habe ein Scala-Skript, das Rohdaten aus S3 entnimmt, verarbeitet und mit @ in HDFS oder sogar S3 schreib Spark-CSV. Ich denke, ich kann mehrere Dateien als Eingabe verwenden, wenn ich @ verwenden möchAWS Machine Learning Tool zum Trainieren eines Vorhersagemodells. Wenn ich jedoch etwas anderes verwenden möchte, ist es wahrscheinlich am besten, wenn ich eine einzelne CSV-Ausgabedatei erhalte.

Zurzeit, da ich nicht @ verwenden möchrepartition (1) Nochcoalesce (1) für Performancezwecke habe ich @ verwendhadoop fs -getmerge zum manuellen Testen, aber da nur der Inhalt der Job-Ausgabedateien zusammengeführt wird, stößt ich auf ein kleines Problem. Ich brauche eine einzelne Kopfzeile in der Datendatei zum Trainieren des Vorhersagemodells.

Wenn ich @ benut.option("header","true") für das spark-csv, dann schreibt es die Header in jede Ausgabedatei und nach dem Zusammenführen habe ich so viele Headerzeilen in den Daten, wie es Ausgabedateien gab. Wenn die Option header jedoch false ist, werden keine Header hinzugefügt.

Nun habe ich eine Option gefunden, mit der ich die Dateien im Scala-Skript mit der Hadoop-API zusammenführen kannFileUtil.copyMerge. Ich habe das in @ versucspark-shell mit dem Code unten.

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

Aber diese Lösung verkettet die Dateien immer noch übereinander und verarbeitet keine Header.Wie kann ich eine Ausgabedatei mit nur einer Kopfzeile erhalten?

Ich habe sogar versucht, @ hinzuzufügdf.columns.mkString(",") als letztes Argument fürcopyMerge, aber dies fügte die Header immer noch mehrmals hinzu, nicht einmal.