Spark Java PCA: Java Heap Space und fehlender Speicherort für das Mischen

Question

Ich versuche, eine PCA auf einem Datenrahmen mit 4,827 Zeilen und 40,107 Spalten durchzuführen, aber ich nehme einen Java-Heap-Space-Fehler und einen fehlenden Ausgabeort für das Shuffle (gemäß der SDTERR-Datei auf Executoren). Der Fehler tritt während der PCA-Phase "treeAggregate at RowMatrix.scala: 122" auf.

Der Cluster

Es handelt sich um einen eigenständigen Cluster mit 16 Worker-Knoten, von denen jeder 1 Executor mit 4 Kernen und 21.504 MB Speicher hat. Der Masterknoten hat 15g Speicher, den ich mit "Java -jar -Xmx15g myapp.jar" gebe. Auch "spark.sql.shuffle.partitions" sind 192 und "spark.driver.maxResultSize" ist 6g.

Vereinfachter Code

df1.persist (From the Storage Tab in spark UI it says it is 3Gb)
df2=df1.groupby(col1).pivot(col2).mean(col3) (This is a df with 4.827 columns and 40.107 rows)
df2.collectFirstColumnAsList
df3=df1.groupby(col2).pivot(col1).mean(col3) (This is a df with 40.107 columns and 4.827 rows)

-----it hangs here for around 1.5 hours creating metadata for upcoming dataframe-----

df4 = (..Imputer or na.fill on df3..)
df5 = (..VectorAssembler on df4..)
(..PCA on df5 with error Missing output location for shuffle..)
df1.unpersist

Ich habe viele Lösungen gesehen und ausprobiert, aber ohne Ergebnis. Darunter:

1. Neu partitionieren von df5 oder df4 auf 16, 64, 192, 256, 1000, 4000 (obwohl die Daten nicht verzerrt aussehen)
2. Ändern der spark.sql.shuffle.partitions in 16, 64, 192, 256, 1000, 4000
3. Verwenden Sie 1 und 2 Kerne pro Executor, um mehr Speicher für jede Aufgabe zu haben.
4. Mit 2 Executoren mit 2 Kernen oder 4 Kernen.
5. Ändern von "spark.memory.fraction" auf 0,8 und "spark.memory.storageFraction" auf 0,4.

Immer der gleiche Fehler! Wie ist es möglich, all diese Erinnerungen wegzublasen? Ist es möglich, dass der df tatsächlich nicht in den Speicher passt? Bitte lassen Sie mich wissen, wenn Sie weitere Informationen oder Druckbildschirme benötigen.

BEARBEITEN 1

Ich habe den Cluster in 2 Spark Worker mit jeweils 1 Executor mit spark.sql.shuffle.partitions = 48 geändert. Jeder Executor hat 115 g und 8 Kerne. Unten ist der Code, in den ich die Datei lade (2,2 GB), jede Zeile in einen dichten Vektor konvertiere und die PCA füttere.

Jede Zeile in der Datei hat dieses Format (4,568 Zeilen mit jeweils 40,107 Doppelwerten):

 "[x1,x2,x3,...]"

und der Code:

Dataset<Row> df1 = sp.read().format("com.databricks.spark.csv").option("header", "true").load("/home/ubuntu/yolo.csv");
StructType schema2 = new StructType(new StructField[] {
                        new StructField("intensity",new VectorUDT(),false,Metadata.empty())
            });
Dataset<Row> df = df1.map((Row originalrow) -> {
                    String yoho =originalrow.get(0).toString();
                    int sizeyoho=yoho.length();
                    String yohi = yoho.substring(1, sizeyoho-1);
                    String[] yi = yohi.split(",");
                    int s = yi.length;
                    double[] tmplist= new double[s];
                    for(int i=0;i<s;i++){
                        tmplist[i]=Double.parseDouble(yi[i]);
                    }
                    
                    Row newrow = RowFactory.create(Vectors.dense(tmplist));
                    return newrow;
            }, RowEncoder.apply(schema2));
PCAModel pcaexp = new PCA()
                    .setInputCol("intensity")
                    .setOutputCol("pcaFeatures")
                    .setK(2)
                    .fit(df);

Der genaue Fehler, den ich auf dem stderr eines der 2 Arbeiter bekomme, ist:

ERROR Executor: Exception in task 1.0 in stage 6.0 (TID 43)
java.lang.OutOfMemoryError
at java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)
at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:117)
at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153)
at org.apache.spark.util.ByteBufferOutputStream.write(ByteBufferOutputStream.scala:41)
at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1877) at java.io.ObjectOutputStream$BlockDataOutputStream.setBlockDataMode(ObjectOutputStream.java:1786)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1189)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:43)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:456) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Und dies ist die Registerkarte "Stufen" von SparkUI:

Und dies ist die Phase, die fehlschlägt (TreeAggregate bei RowMatrix.scala: 122):

BEARBEITEN 2

BEARBEITEN 3

Ich habe die ganze Datei gelesen, aber nur 10 Werte aus jeder Zeile genommen und den dichten Vektor erstellt. Ich bekomme immer noch den gleichen Fehler! Ich habe einen Master mit 235 g Ram und 3 Arbeitern (je 1 Executor mit 4 Kernen) und 64 g Ram pro Executor. Wie könnte das passieren? (Vergessen Sie nicht, dass die Gesamtgröße der Datei nur 2,3 GB beträgt!)

Dataset<Row> df1 = sp.read().format("com.databricks.spark.csv").option("header", "true").load("/home/ubuntu/yolo.csv");

StructType schema2 = new StructType(new StructField[] {
                        new StructField("intensity",new VectorUDT(),false,Metadata.empty())
            });
Dataset<Row> df = df1.map((Row originalrow) -> {
                    String yoho =originalrow.get(0).toString();
                    int sizeyoho=yoho.length();
                    String yohi = yoho.substring(1, sizeyoho-1);
                    String[] yi = yohi.split(",");//this string array has all 40.107 values
                    int s = yi.length;
                    double[] tmplist= new double[s];
                    for(int i=0;i<10;i++){//I narrow it down to take only the first 10 values of each row
                        tmplist[i]=Double.parseDouble(yi[i]);
                    }
                    Row newrow = RowFactory.create(Vectors.dense(tmplist));
                    return newrow;
            }, RowEncoder.apply(schema2));
      
PCAModel pcaexp = new PCA()
                    .setInputCol("intensity")
                    .setOutputCol("pcaFeatures")
                    .setK(2)
                    .fit(df);

Answer 1

Der "Fehlende Ausgabeort für Shuffle" tritt auf, wenn Ihre Spark-Anwendung große Shuffle-Phasen ausführt, versucht, eine große Datenmenge unter den Ausführenden neu zuzuweisen, und es gibt einige Probleme in Ihrem Clusternetzwerk.

Spark sagt, dass du irgendwann kein Gedächtnis mehr hast. Sie führen Transformationen durch, die unterschiedliche Phasen erfordern und auch Speicher verbrauchen. Außerdem behalten Sie zuerst den Datenrahmen bei, und Sie sollten die Speicherebene überprüfen, da es möglich ist, dass Sie im Speicher verbleiben.

Sie verketten mehrere Spark-weite Transformationen: Wenn Sie beispielsweise die erste Pivot-Phase ausführen, erstellt Spark eine Phase und führt ein Shuffle durch, um Ihre Spalte zu gruppieren. Möglicherweise haben Sie einen Datenversatz, und es gibt Executoren, die viel mehr Speicher verbrauchen als andere Der Fehler kann in einem von ihnen auftreten.

Neben den Datenrahmen-Transformationen konvertiert der PCA-Schätzer den Datenrahmen in eine RDD, wodurch der Speicher zur Berechnung der Kovarianzmatrix wesentlich größer wird, und er arbeitet mit dichten Darstellungen von Breeze-Matrizen von NxN-Elementen, die nicht verteilt sind . Zum Beispiel wird die SVD mit Breeze erstellt. Das hat einen der Testamentsvollstrecker stark unter Druck gesetzt.

Vielleicht können Sie den resultierenden Datenrahmen in HDFS (oder was auch immer) speichern und die PCA einer anderen Spark-Anwendung ausführen.

Das Hauptproblem. Was Sie haben, ist, dass der Algorithmus vor der SVD die Grammian-Matrix berechnen muss und ein treeAggregate von RDD verwendet. Dadurch wird eine sehr große Doppelmatrix erstellt, die an den Treiber gesendet wird. Es liegt ein Fehler vor, weil Ihr Treiber nicht über genügend Speicher verfügt. Sie müssen den Treiberspeicher drastisch erhöhen. Sie haben Netzwerkfehler. Wenn ein Executor die Verbindung verliert und der Job abstürzt, versucht er nicht, ihn erneut auszuführen.

Persönlich würde ich versuchen, die PCA direkt in Breeze (oder Smile) im Treiber durchzuführen. Ich meine, das RDD-Feld zu sammeln, da der Datensatz ziemlich kleiner als die Covarianze-Matrix ist, und manuell mit einer Float-Darstellung.

Code zum Berechnen der PCA nur mit Breeze, weder Spark noch TreeAgregation:

import breeze.linalg._
import breeze.linalg.svd._

object PCACode {
  
  def mean(v: Vector[Double]): Double = v.valuesIterator.sum / v.size

  def zeroMean(m: DenseMatrix[Double]): DenseMatrix[Double] = {
    val copy = m.copy
    for (c <- 0 until m.cols) {
      val col = copy(::, c)
      val colMean = mean(col)
      col -= colMean
    }
    copy
  }

  def pca(data: DenseMatrix[Double], components: Int): DenseMatrix[Double] = {
    val d = zeroMean(data)
    val SVD(_, _, v) = svd(d.t)
    val model = v(0 until components, ::)
    val filter = model.t * model
    filter * d
  }
  
  def main(args: Array[String]) : Unit = {
    val df : DataFrame = ???

    /** Collect the data and do the processing. Convert string to double, etc **/
    val data: Array[mutable.WrappedArray[Double]] =
      df.rdd.map(row => (row.getAs[mutable.WrappedArray[Double]](0))).collect()

    /** Once you have the Array, create the matrix and do the PCA **/
    val matrix = DenseMatrix(data.toSeq:_*)
    val pcaRes = pca(matrix, 2)

    println("result pca \n" + pcaRes)
  }
}

Dieser Code führt die PCA im Treiber durch, überprüft den Speicher. Wenn es abstürzt, kann dies mit einer Float-Präzision geschehen.