Spark Java PCA: espace de tas Java et emplacement de sortie manquant pour la lecture aléatoire

Question

J'essaie de faire un PCA sur une trame de données avec 4,827 lignes et 40,107 colonnes mais je prends une erreur d'espace de tas Java et un emplacement de sortie manquant pour le shuffle (selon le fichier sdterr sur les exécuteurs). L'erreur se produit lors de l' étape "treeAggregate at RowMatrix.scala: 122" de PCA.

Le cluster

Il s'agit d'un cluster autonome avec 16 nœuds de travail, chacun ayant 1 exécuteur avec 4 cœurs et 21,504 Mo de mémoire. Le nœud maître a 15g de mémoire que je donne avec "Java -jar -Xmx15g myapp.jar". De plus, "spark.sql.shuffle.partitions" vaut 192 et "spark.driver.maxResultSize" vaut 6g.

Code simplifié

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

J'ai vu et essayé de nombreuses solutions mais sans aucun résultat. Parmi eux:

1. Re-partitionnement du df5 ou du df4 en 16, 64, 192, 256, 1000, 4000 (bien que les données ne semblent pas faussées)
2. Modification des partitions spark.sql.shuffle.partitions en 16, 64, 192, 256, 1000, 4000
3. Utiliser 1 et 2 cœurs par exécuteur pour avoir plus de mémoire pour chaque tâche.
4. Avoir 2 exécuteurs avec 2 cœurs ou 4 cœurs.
5. Changement de "spark.memory.fraction" à 0.8 et "spark.memory.storageFraction" à 0.4.

Toujours la même erreur! Comment est-il possible de souffler tout ce souvenir ?? Est-il possible que le df ne rentre pas dans la mémoire? S'il vous plaît laissez-moi savoir si vous avez besoin d'autres informations ou écrans d'impression.

MODIFIER 1

J'ai changé le cluster en 2 ouvriers Spark avec 1 exécuteur chacun avec spark.sql.shuffle.partitions = 48. Chaque exécuteur a 115g et 8 cœurs. Voici le code où je charge le fichier (2,2 Go), convertis chaque ligne en un vecteur dense et alimente le PCA.

Chaque ligne du fichier a ce format (4,568 lignes avec 40,107 valeurs doubles chacune):

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

et le 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);

L'erreur exacte que j'obtiens sur le stderr de l'un des 2 travailleurs est:

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)

Et voici l'onglet Stages de SparkUI:

Et c'est l'étape qui échoue (TreeAggregate at RowMatrix.scala: 122):

MODIFIER 2

MODIFIER 3

J'ai lu le fichier entier mais en prenant seulement 10 valeurs de chaque ligne et en créant le vecteur dense. Je reçois toujours la même erreur! J'ai un maître avec 235g de RAM et 3 ouvriers (1 exécuteur avec chacun 4 cœurs) et 64g de RAM par exécuteur. Comment cela pourrait-il se produire? (N'oubliez pas que la taille totale du fichier n'est que de 2,3 Go!)

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

« L' emplacement de sortie manquant pour le shuffle» se produit lorsque votre application Spark effectue de grandes étapes de shuffle, elle essaie de réallouer une énorme quantité de données entre les exécuteurs et il y a des problèmes dans votre réseau de cluster.

Spark dit que vous n'avez pas de mémoire à un moment donné. Vous effectuez des transformations qui nécessitent différentes étapes et qui consomment également de la mémoire. En outre, vous persistez d'abord dans la trame de données et vous devez vérifier le niveau de stockage, car il est possible que vous persistiez en mémoire.

Vous enchaînez plusieurs transformations étendues à Spark: lors de la première étape de pivot, par exemple, Spark crée une étape et effectue un mélange pour regrouper votre colonne et peut-être avez-vous des données biaisées et il y a des exécuteurs qui consomment beaucoup plus de mémoire que d'autres, et peut-être l'erreur peut se produire dans l'un d'entre eux.

Outre les transformations Dataframe, l'estimateur PCA convertit la dataframe en RDD augmentant beaucoup plus la mémoire pour calculer la matrice covarianze, et il fonctionne avec des représentations denses de matrices Breeze d'éléments NxN qui ne sont pas distribués . Par exemple, le SVD est fait avec Breeze. Cela a mis beaucoup de pression sur l'un des exécuteurs testamentaires.

Vous pouvez peut-être enregistrer la trame de données résultante dans HDFS (ou autre) et faire de la PCA une autre application Spark.

Le problème principal. ce que vous avez, c'est qu'avant de SVD, l'algorithme doit calculer la matrice de Grammian et il utilise un treeAggregate de RDD. Cela crée une très grande matrice Double qui sera envoyée au pilote, et il y a l'erreur parce que votre pilote n'a pas assez de mémoire. Vous devez augmenter considérablement la mémoire du pilote. Vous avez des erreurs de réseau, si un exécuteur perd la connexion, le travail se bloque, il n'essaye pas de se réexécuter.

Personnellement, j'essaierais de faire le PCA directement dans Breeze (ou Smile) dans le pilote, je veux dire, collecter le champ RDD car l'ensemble de données est assez plus petit que la matrice covarianze et le faire manuellement avec une représentation Float.

Code pour calculer la PCA uniquement avec Breeze, ni Spark ni 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)
  }
}

Ce code fera le PCA dans le pilote, vérifier la mémoire. S'il plante, cela pourrait être dû à une précision Float.