Spark程序性能——GC&任务反序列化&并发执行

Question

我有一个4台机器的集群，1台master，3台worker，每台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) 

问题：

1. 由于我的 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时间和任务反序列化时间？

2. 我必须对 h[1-24] 进行并行计算，而不是顺序计算。我尝试了未来，类似：

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

问题是，这里的 Future 仅意味着作业几乎同时提交给调度程序，而不是它们最终被调度并同时运行。这里使用 Future 根本不会提高性能。

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

Answer 1

您可以尝试以下几件事：

1. 不用pmod(idx, 24)再重新计算了。相反，您可以简单地计算一次：

   import org.apache.spark.sql.functions.{pmod, lit}
   
   val myDfWithBuckets = myDF.withColumn("bucket", pmod($"idx", lit(24)))
   

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

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

3. 如果可以的话，只缓存您实际需要的列，而不是每次都进行投影。

4. 我不清楚histogram方法的来源是什么（您是否转换为RDD[Double]并使用DoubleRDDFunctions.histogram？）以及参数是什么，但如果您想同时计算所有直方图，您可以尝试存储groupBy并应用一次直方图，例如使用histogram_numericUDF：

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