有没有办法优化 spark sql 代码?

Posted

技术标签:

【中文标题】有没有办法优化 spark sql 代码?【英文标题】:is there a way to optimize spark sql code? 【发布时间】:2016-02-03 10:35:30 【问题描述】:

更新:

我正在使用 spark sql 1.5.2。尝试读取许多 parquet 文件并过滤和聚合行 - 在我的 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 内存,scala 输出让我认为计算只在一个线程中运行:

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

当我尝试 count() 而不是 first() 时,看起来两个线程正在执行计算。这仍然比我预期的要少,因为有大约 30 个文件可以并行处理

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

我正在启动 spark 控制台,其中 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

有没有办法优化这段代码? 谢谢

【问题讨论】:

由于您使用first 在某些时候只看到一个活动线程,这并不奇怪。拨打count时问题是否依然存在? 你是对的,当我调用 count 时,它看起来在两个线程中运行,这仍然比我预期的要少一些,因为有 30 个 parquet 文件可以并行映射,并且可以有尽可能多的reducer 作为 pid 和 date 的唯一值。我也会更新我的问题 您还应该提供配置详细信息(集群管理器,spark.executor.cores,默认并行度)。另请检查分区数(logins_12.rdd.partitions.size)。 再次对,抱歉。更新的问题 【参考方案1】:

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

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

【讨论】:

以上是关于有没有办法优化 spark sql 代码?的主要内容,如果未能解决你的问题,请参考以下文章

Spark sql 查询优化

有spark大佬知道下面这个代码哪里可以优化吗?

spark如何优化设置

Spark SQL项目中的优化思路

Hive和Spark SQL优化

Hive和Spark SQL优化