Spark Java PCA: Java Heap Space and Missing output location for shuffle

Question

Tento fazer um PCA em um dataframe com 4.827 linhas e 40.107 colunas, mas recebo um erro de espaço de heap Java e um local de saída ausente para embaralhamento (de acordo com o arquivo sdterr nos executores). O erro ocorre durante o estágio "treeAggregate at RowMatrix.scala: 122" do PCA.

O cluster

É um cluster autônomo com 16 nós de trabalho, cada um com 1 executor com 4 núcleos e 21,504 MB de memória. O nó mestre tem 15g de memória que forneço com "Java -jar -Xmx15g myapp.jar". Além disso, "spark.sql.shuffle.partitions" são 192 e "spark.driver.maxResultSize" é 6g.

Código simplificado

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á vi e tentei muitas soluções, mas sem resultado. Entre eles:

1. Reparticionamento do df5 ou df4 para 16, 64, 192, 256, 1000, 4000 (embora os dados não pareçam distorcidos)
2. Alterando spark.sql.shuffle.partitions para 16, 64, 192, 256, 1000, 4000
3. Usando 1 e 2 núcleos por executor para ter mais memória para cada tarefa.
4. Ter 2 executores com 2 núcleos ou 4 núcleos.
5. Alterar "spark.memory.fraction" para 0.8 e "spark.memory.storageFraction" para 0.4.

Sempre o mesmo erro! Como é possível explodir toda essa memória ?? É possível que o df realmente não se ajuste na memória? Por favor, deixe-me saber se você precisar de qualquer outra informação ou printcreens.

EDITAR 1

Mudei o cluster para 2 trabalhadores spark com 1 executor cada com spark.sql.shuffle.partitions = 48. Cada executor possui 115g e 8 núcleos. Abaixo está o código onde carrego o arquivo (2.2Gb), converto cada linha em um vetor denso e alimento o PCA.

Cada linha do arquivo tem este formato (4.568 linhas com 40.107 valores duplos cada):

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

e o código:

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

O erro exato que recebo no stderr de um dos 2 trabalhadores é:

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)

E esta é a guia Estágios do SparkUI:

E este é o estágio que falha (TreeAggregate em RowMatrix.scala: 122):

EDITAR 2

EDITAR 3

Eu li o arquivo inteiro, mas pegando apenas 10 valores de cada linha e criando o vetor denso. Ainda obtenho o mesmo erro! Eu tenho um mestre com 235g de Ram e 3 trabalhadores (1 executor cada com 4 núcleos) e 64g de Ram por executor. Como isso pode estar acontecendo? (Não se esqueça de que o tamanho total do arquivo é de apenas 2,3 Gb!)

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

O "Local de saída ausente para embaralhamento" ocorre quando seu aplicativo Spark realiza grandes estágios de embaralhamento, ele tenta realocar uma grande quantidade de dados entre os executores e há alguns problemas em sua rede de cluster.

Spark diz que você não tem memória em alguma fase. Você está fazendo transformações que requerem diferentes estágios e também consomem memória. Além disso, primeiro você persiste o dataframe e deve verificar o nível de armazenamento, pois é possível que você esteja persistindo na memória.

Você está encadeando várias transformações amplas do Spark: fazendo o primeiro estágio de pivô, por exemplo, o Spark cria um estágio e executa uma ordem aleatória para agrupar para sua coluna e talvez você tenha distorção de dados e existam executores que consomem muito mais memória do que outros, e talvez o erro pode acontecer em um deles.

Além das transformações do Dataframe, o estimador PCA converte o dataframe para um RDD aumentando muito mais a memória para calcular a matriz de covarianze, e trabalha com representações densas de matrizes Breeze de elementos NxN que não são distribuídos . Por exemplo, o SVD é feito com Breeze. Isso colocou muita pressão em um dos executores.

Talvez você possa salvar o dataframe resultante em HDFS (ou qualquer outro) e fazer o PCA em outro aplicativo Spark.

O problema principal. que você tem é que antes de SVD o algoritmo precisa computar a matriz Grammian e usa um treeAggregate de RDD. Isso cria uma matriz dupla muito grande que será enviada para o driver, e ocorre o erro porque o driver não tem memória suficiente. Você precisa aumentar drasticamente a memória do driver. Você tem erros de rede, se um executor perder a conexão, o job trava, ele não tenta reexecutar.

Pessoalmente, eu tentaria fazer o PCA diretamente no Breeze (ou Smile) no driver, quer dizer, coletar o campo RDD porque o conjunto de dados é bem menor que a matriz covarianze e fazer manualmente com uma representação Float.

Código para calcular o PCA apenas com Breeze, nem Spark nem 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)
  }
}

Este código vai fazer o PCA no driver, verifique a memória. Se ele falhar, pode ser devido a uma precisão de Float.