Ich bin ziemlich neu in Spark und verteilten Computing. Es ist alles sehr einfach, eine CSV-Datei oder eine Textdatei zu laden, die in Ihren Speicher passen. Aber hier habe ich ein echtes Szenario und ich finde es schwierig, den Ansatz zu verstehen.Spark - Laden eines Verzeichnisses mit 500G Daten
Ich versuche, in S3 auf etwa 500 G Daten zuzugreifen, diese besteht aus Zip-Dateien. Da dies ZIP-Dateien sind, verwende ich ZipFileInputFormat wie detailliert here. Es stellt sicher, dass die Dateien nicht über Partitionen verteilt werden.
Hier ist meine Code val sc = new SparkContext (conf)
val inputLocation = args(0)
val emailData = sc.newAPIHadoopFile(inputLocation, classOf[ZipFileInputFormat], classOf[Text], classOf[BytesWritable]);
val filesRDD = emailData.filter(_._1.toString().endsWith(".txt")).map(x => new String(x._2.getBytes))
Dies läuft auf einer Eingabe von wenigen 100mb in Ordnung. Aber sobald es das Speicherlimit meines Clusters überschreitet, erhalte ich das Problem out ofMemory.
Was ist der richtige Weg, dieses Problem anzugehen? - sollte ich eine RDD für jede Zip-Datei erstellen und die Ausgabe in eine Datei speichern und später alle Ausgaben in eine separate RDD laden? - Gibt es eine Möglichkeit, das Basisverzeichnis in Spark-Kontext zu laden und zu partitionieren
Ich habe einen HDP-Cluster mit 5 Knoten und einem Master mit jeweils 15G Speicher.
Alle Antworten/Zeiger/Links werden sehr geschätzt
Ich stimme zu Zip-Dateien sind nicht teilbar, aber diese sind jeweils maximal 1GB. Ich habe festgestellt, dass keine Partitionen erstellt wurden, und es wurde eine Partition für jede Zip-Datei erstellt. Also wird die Partitionsgröße größer als der Treiber-/Executor-Speicher und daher das Problem? – krc
Not splitable => Eine Partition pro Datei ist. Speicherprobleme sind entweder überschreiten Grenzen oder GC. –