Spark Java PCA: Java Heap Space ve karıştırma için Eksik çıktı konumu

Question

4.827 satırlı ve 40.107 sütunlu bir veri çerçevesi üzerinde bir PCA yapmaya çalışıyorum, ancak bir Java yığın alanı hatası alıyorum ve karıştırma için eksik çıktı konumu (yürütücülerdeki sdterr dosyasına göre). Hata, PCA'nın "RowMatrix.scala'daki treeAggregate: 122" aşamasında meydana gelir.

Küme

Her biri 4 çekirdekli 1 yürütücüye ve 21.504mb belleğe sahip 16 çalışan düğümü olan bağımsız bir kümedir. Ana düğümde "Java -jar -Xmx15g myapp.jar" ile verdiğim 15g bellek var. Ayrıca "spark.sql.shuffle.partitions" 192 ve "spark.driver.maxResultSize" 6g'dir.

Basitleştirilmiş kod

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

Pek çok çözüm gördüm ve denedim ama sonuçsuz. Onların arasında:

1. Df5 veya df4'ü 16, 64, 192, 256, 1000, 4000'e yeniden bölümleme (veriler çarpık görünmese de)
2. Spark.sql.shuffle.partitions'ı 16, 64, 192, 256, 1000, 4000 olarak değiştirme
3. Her görev için daha fazla belleğe sahip olmak için yürütücü başına 1 ve 2 çekirdek kullanmak.
4. 2 çekirdekli veya 4 çekirdekli 2 yürütücüye sahip.
5. "Spark.memory.fraction" 0,8 ve "spark.memory.storageFraction" 0,4 olarak değiştiriliyor.

Hep aynı hata! Tüm bu anıları yok etmek nasıl mümkün olabilir? Df gerçekten belleğe sığmıyor olabilir mi? Başka bir bilgiye veya baskı ekranına ihtiyacınız olursa lütfen bana bildirin.

DÜZENLE 1

Kümeyi, her biri spark.sql.shuffle.partitions = 48 olan 1 yürütücüye sahip 2 spark işçisine değiştirdim. Her yürütücünün 115g ve 8 çekirdeği vardır. Dosyayı (2.2Gb) yüklediğim, her satırı yoğun bir vektöre dönüştürdüğüm ve PCA'yı beslediğim kod aşağıdadır.

Dosyadaki her satır şu biçime sahiptir (her biri 40.107 çift değer içeren 4.568 satır):

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

ve kod:

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

2 işçiden birinin stderrinde aldığım kesin hata şudur:

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)

Ve işte SparkUI'nin Aşamalar Sekmesi:

Ve bu, başarısız olan Aşama'dır (TreeAggregate, RowMatrix.scala: 122):

DÜZENLEME 2

DÜZENLEME 3

Dosyanın tamamını okudum ama her satırdan sadece 10 değer alıp yoğun vektör oluşturdum. Ben hala aynı hatayı alıyorum! 235g Ram ve 3 işçi (her biri 4 çekirdekli 1 uygulayıcı) ve uygulayıcı başına 64g Ram olan bir ustam var. Bu nasıl olabilir? (Dosyanın toplam boyutunun sadece 2.3Gb olduğunu unutmayın!)

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

"Shuffle için Eksik çıkış konumu" senin Spark uygulama büyük shuffle aşamalarını yapmak, bunun uygulayıcıları arasında veri büyük miktarda yeniden tahsis etmeye çalışır ve küme ağında bazı sorunlar var oluşur.

Spark, bir aşamada hafızanız olmadığını söylüyor. Farklı aşamalar gerektiren dönüşümler yapıyorsunuz ve onlar da hafızayı tüketiyor. Ayrıca, önce veri çerçevesini devam ettirirsiniz ve depolama düzeyini kontrol etmelisiniz, çünkü bellekte devam ediyor olmanız olasıdır.

Birkaç Spark geniş dönüşümü zincirliyorsunuz: ilk pivot aşamasını yapmak, örneğin, Spark bir sahne oluşturur ve sütununuz için bir gruplandırma gerçekleştirir ve belki veri çarpıklığınız vardır ve diğerlerinden çok daha fazla bellek tüketen yürütücüler vardır ve belki hata bunlardan birinde olabilir.

Veri Çerçevesi dönüşümlerinin yanı sıra, PCA tahmincisi veri çerçevesini bir RDD'ye dönüştürür ve kovaryans matrisini hesaplamak için belleği çok daha fazla arttırır ve dağıtılmamış NxN elemanlarının Breeze matrislerinin yoğun temsilleriyle çalışır . Örneğin, SVD Breeze ile yapılır. Bu, uygulayıcılardan birine çok fazla baskı uyguladı.

Belki elde edilen veri çerçevesini HDFS'ye (veya her neyse) kaydedebilir ve PCA'yı başka bir Spark uygulaması yapabilirsiniz.

Ana problem. Elinizdeki, de SVD'den önce algoritmanın Grammian Matrix'i hesaplaması gerekiyor ve RDD'den bir treeAggregate kullanıyor. Bu, sürücüye gönderilecek çok büyük bir Çift matris yaratır ve sürücünüzün hafızası yeterli olmadığı için bir hata vardır. Sürücü belleğini önemli ölçüde artırmanız gerekir. Ağ hatalarınız var, bir yürütücü bağlantıyı kaybederse iş çökerse, yeniden yürütmeyi denemiyor.

Kişisel olarak, PCA'yı doğrudan sürücüde Breeze'de (veya Smile'da) yapmaya çalışırım, yani RDD alanını toplar, çünkü veri seti kovaryan matrisinden oldukça küçüktür ve Float gösterimi ile manuel olarak yapılır.

PCA'yı yalnızca Breeze ile hesaplamak için kodlayın, ne Spark ne de 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)
  }
}

Bu kod sürücüde PCA yapacak, hafızayı kontrol edecektir. Çarparsa, Float öncülüğüyle yapılabilir.