scala – Spark程序性能 – GC和任务反序列化和并发执行

2023年3月13日 222次阅读

我有一个由4台机器组成的集群,1台主机和3台工人,每台机器有128G内存和64个内核.我在独立模式下使用Spark 1.5.0.我的程序使用JDBC从Oracle表中读取数据,然后执行ETL,操作数据,并执行机器学习任务,如k-means.

我有一个DataFrame(myDF.cache()),它与其他两个DataFrame连接结果,并缓存. DataFrame包含2700万行,数据大小约为1.5G.我需要过滤数据并计算24个直方图,如下所示：

val h1 = myDF.filter("pmod(idx, 24) = 0").select("col1").histogram(arrBucket) 
val h2 = myDF.filter("pmod(idx, 24) = 1").select("col1").histogram(arrBucket) 
// ...... 
val h24 = myDF.filter("pmod(idx, 24) = 23").select("col1").histogram(arrBucket)

问题：

>由于我的DataFrame被缓存,我希望过滤器,选择和直方图非常快.但是,每次计算的实际时间约为7秒,这是不可接受的.在UI中,它显示GC时间需要5秒,任务反序列化时间需要4秒.我尝试过不同的JVM参数但无法进一步改进.现在我正在使用

-Xms25G -Xmx25G -XX:MaxPermSize=512m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 \
-XX:ParallelGCThreads=32 \
-XX:ConcGCThreads=8 -XX:InitiatingHeapOccupancyPercent=70

令我困惑的是,与可用内存相比,数据的大小无关紧要.为什么每次过滤/选择/直方图运行时GC都会启动？有没有办法减少GC时间和任务反序列化时间？

>我必须为h [1-24]进行并行计算,而不是顺序计算.我尝试过Future,类似于：

import scala.concurrent.{Await, Future, blocking} 

import scala.concurrent.ExecutionContext.Implicits.global 

val f1  = Future{myDF.filter("pmod(idx, 24) = 1").count} 
val f2  = Future{myDF.filter("pmod(idx, 24) = 2").count} 
val f3  = Future{myDF.filter("pmod(idx, 24) = 3").count} 

val future = for {c1 <- f1; c2 <- f2; c3 <- f3} yield { 
  c1 + c2 + c3 
} 

val summ = Await.result(future, 180 second)

问题在于,Future只是意味着工作几乎同时提交给调度程序,而不是它们最终被安排并同时运行.这里使用的未来根本不会提高性能.

如何使24个计算作业同时运行？

最佳答案 你可以试试几件事：

>不要再重新计算pmod(idx,24).相反,你可以简单地计算一次：

import org.apache.spark.sql.functions.{pmod, lit}

val myDfWithBuckets = myDF.withColumn("bucket", pmod($"idx", lit(24)))

>使用SQLContext.cacheTable而不是缓存.它使用压缩列存储来存储表,该存储可用于仅访问所需的列,并且如Spark SQL and DataFrame Guide中所述“将自动调整压缩以最小化内存使用和GC压力”.

myDfWithBuckets.registerTempTable("myDfWithBuckets")
sqlContext.cacheTable("myDfWithBuckets")

>如果可以,只缓存实际需要的列,而不是每次都进行投影.
>我不清楚直方图方法的来源是什么(你转换为RDD [Double]并使用DoubleRDDFunctions.histogram？)和什么是参数但如果你想同时计算所有直方图你可以尝试groupBy存储桶并应用直方图一次,例如使用histogram_numeric UDF：

import org.apache.spark.sql.functions.callUDF

val n: Int = ???

myDfWithBuckets
  .groupBy($"bucket")
  .agg(callUDF("histogram_numeric", $"col1", lit(n)))

如果使用预定义范围,则可以使用自定义UDF获得类似的效果.

笔记

>如何提取histogram_numeric计算的值？首先让我们创建一个小助手

import org.apache.spark.sql.Row

def extractBuckets(xs: Seq[Row]): Seq[(Double, Double)] =
  xs.map(x => (x.getDouble(0), x.getDouble(1)))

现在我们可以使用模式匹配来映射如下：

import org.apache.spark.rdd.RDD

val histogramsRDD: RDD[(Int, Seq[(Double, Double)])] = histograms.map{
  case Row(k: Int, hs: Seq[Row @unchecked]) => (k, extractBuckets(hs)) }
点赞