有没有办法优化Spark SQL代码

Question

>更新：

我正在使用火花sql 1.5.2。尝试读取许多镶木地板文件并过滤和聚合行 - 在我的 hdfs 中，~30 个文件中存储了 ~35M 的行，处理时间超过 10 分钟

val logins_12 = sqlContext.read.parquet("events/2015/12/*/login")
val l_12 = logins_12.where("event_data.level >= 90").select(
    "pid", 
    "timestamp", 
    "event_data.level" 
    ).withColumn("event_date", to_date(logins_12("timestamp"))).drop("timestamp").toDF("pid",  "level", "event_date").groupBy("pid", "event_date").agg(Map("level"->"max")).toDF("pid", "event_date", "level")
l_12.first()   

我的 Spark 在两个节点集群中运行，每个集群都有 8 个核心和 16Gb RAM，Scala 输出让我明白计算只在一个线程中运行：

scala> x.first()
[Stage 1:=======>                                               (50 + 1) / 368]

当我尝试 count（） 而不是 first（） 时，看起来两个线程正在进行计算。 这仍然比我预期的要少，因为有 ~30 个文件可以并行处理

scala> l_12.count()   
[Stage 4:=====>                                                  (34 + 2) / 368]

我正在启动 Spark 控制台，在 yarn-client 模式下，执行器 14g 和驱动程序为 4g

./bin/spark-shell -Dspark.executor.memory=14g -Dspark.driver.memory=4g --master yarn-client

我对Spark的默认配置：

spark.executor.memory              2g
spark.logConf                      true
spark.eventLog.dir                 maprfs:///apps/spark
spark.eventLog.enabled             true
spark.sql.hive.metastore.sharedPrefixes  com.mysql.jdbc,org.postgresql,com.microsoft.sqlserver,oracle.jdbc,com.mapr.fs.shim.LibraryLoader,com.mapr.security.JNISecurity,com.mapr.fs.jni
spark.executor.extraClassPath
spark.yarn.historyServer.address  http://test-01:18080

RDD 有 200 个分区

scala> logins_12.rdd.partitions.size
res2: Int = 368
scala> l_12.rdd.partitions.size
res0: Int = 200

有没有办法优化此代码？谢谢

Answer 1

这两种行为或多或少都是意料之中的。Spark 相当懒惰，它不仅不会执行转换，除非您触发操作，而且如果输出不需要，还可以跳过任务。由于first只需要一个元素，因此它只能计算一个分区。这很可能是您在某个时候只看到一个正在运行的线程的原因。

关于第二个问题，这很可能是配置问题。假设 YARN 配置没有任何问题（我不使用 YARN，但看起来yarn.nodemanager.resource.cpu-vcores可能是问题的根源），这很可能是 Spark 默认值的问题。正如您在配置指南中读到的那样spark.executor.cores Yarn 默认设置为 1。两个工作线程提供两个运行线程。