Speichern einer großen Datei überschreitet frameLimit

Question

Ich versuche, eine große Textdatei von ca. 5GB

sc.parallelize(cfile.toString() 
    .split("\n"), 1) 
    .saveAsTextFile(new Path(path+".cs", "data").toUri.toString) 

aber ich bekomme immer wieder

java.io.IOException: Broken pipe 
    at sun.nio.ch.FileDispatcherImpl.write0(Native Method) 
    at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:47) 
    at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) 
    at sun.nio.ch.IOUtil.write(IOUtil.java:65) 
... 
org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 6 
    at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:542) 
    at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:538) 
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) 
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) 

Ich bin jetzt hier seit Ewigkeiten fest. Kann mir jemand hier helfen und erklären, wie ich cfile als Textdatei speichern kann?

---

Eigenständig/lokal/Garn-Cluster?

• Yarn Cluster

Speicher/Cores Einstellungen?

• 1,8 TB
• 285 Cores

Anzahl der Partitionen?

• ich zur Zeit die Anzahl der Partitionen 1 bin Einstellung: zum Einstellen der Anzahl der Partitionen

Die betreffenden Codezeilen:

val model = word2vec 
    .setMinCount(minCount.asInstanceOf[Int]) 
    .setVectorSize(arguments.getVectorSize) 
    .setWindowSize(arguments.getContextWindowSize) 
    .setNumPartitions(numW2vPartitions) 
    .setLearningRate(learningRate) 
    .setNumIterations(arguments.getNumIterations) 
    .fit(wordSequence) 

Funken einreichen Argumente:

spark-submit --master yarn 
      --deploy-mode cluster 
      --driver-memory 20G 
      --num-executors 5 
      --executor-cores 8 
      --driver-java-options "-Dspark.akka.frameSize=2000" 
      --executor-memory 20G --class 

Answer 1

Eigenständig/lokal/Garn-Cluster? Speicher/Kerne Einstellungen? Anzahl der Partitionen?

Ihr Fehler wahrscheinlich Symptom, dass einer der Arbeiter sind weg (OOM Killer könnte es getötet oder es habe einige OOM-Fehler)

Ich bin mir nicht sicher, warum Sie dies tun: cfile.toString(). split ("\ n") - von diesem verstehe ich, dass Sie alle 5GB Inhalt im Speicher halten und versuchen, es zu parallelisieren? Klar ist es nicht optimal. Ein anderes Problem, das relevant sein könnte - wenn Ihr Treiber alle 5GB irgendwie im Speicher halten kann, aber trotzdem alle Netzwerkebenen zwischen Treiber-Arbeitern diese Datenmenge nicht mögen - also mein Rat, es in Partitionen aufzuteilen.

Stattdessen können Sie die Datei mit sc.textFile (..) lesen und dann in Ihrem neuen Pfad speichern. Sie können auch die Anzahl der Partitionen (Teile) Ihrer Textdatei mit sc.textFile (..) .partition (100) steuern.