如何优化大数据框上的 spark sql 操作？答案

【问题标题】：How to optimize spark sql operations on large data frame?如何优化大数据框上的 spark sql 操作？
【发布时间】：2017-07-10 14:13:33
【问题描述】：

我有一个大型配置单元表（约 90 亿条记录和约 45GB 的兽人格式）。我正在使用 spark sql 对表进行一些分析。但是对此进行任何操作需要花费太多时间。只需对输入数据框本身进行计数就需要大约 11 分钟才能完成。仅任何一列的 min、max 和 avg 都需要超过一个半小时才能完成。

我正在开发一个资源有限的集群（因为它是唯一可用的集群），共有 9 个执行程序，每个执行程序有 2 个核心，每个执行程序有 5GB 内存，分布在 3 个物理节点上。

有什么方法可以优化这一点，比如将在同一集群上执行每列上的所有聚合函数的时间减少到至少 30 分钟以内，或者增加我的资源是唯一的方法？我个人不太热衷于这样做。我遇到的一种加快数据帧操作的解决方案是缓存它们。但在我的情况下，我认为这不是一个可行的选择。

我遇到的所有现实世界场景都使用巨大的集群来处理这种负载。

感谢任何帮助。我在独立模式下使用 spark 1.6.0 和 kryo 序列化程序。

【问题讨论】：

您可以添加一些代码以及您使用什么配置来运行该作业？表是如何分区的（如果有的话）？
@morfious902002 我已经提到了我的环境。我运行 dataframe.count() 或 hiveContext.sql("select max(column),min(column),avg(column) from table" ) 并且 hive 表没有分区，但它使用包含参数的列进行存储名字。
尝试 3 个执行器，每个执行器有 3-5 个内核和 15 GB 内存。对于缓存，您可以尝试 .persist(StorageLevel.MEMORY_AND_DISK_SER())
@morfious902002 我已经尝试在 memeor_and_disk_ser 上进行持久化，它本身需要大约 90 分钟才能持久化，并且由于 ram 为 45GB，大部分都进入了磁盘。并没有太大帮助。尽管它可以完全适合内存，但它有助于减少数据量。我会尝试将执行者与更大的执行者结合起来，谢谢。但是你能解释一下它有什么帮助吗？
与物理机相比，您的小型执行器太多。这反过来会导致更多的开销。此外，在使用这些功能之前，请尝试进行过滤，选择您需要的列（如果尚未完成）。

【解决方案1】：

sparkSQL 中有一些很酷的功能，例如：

集群方式/分发方式/排序方式

Spark 允许您使用类似 SQL 的语言 - HiveQL 编写查询。 HiveQL 让您可以控制数据的分区，就像我们可以在 SparkSQL 查询中使用它一样。

分发者

在spark中，Dataframe被某个表达式分区，这个表达式相等的所有行都在同一个分区上。

SET spark.sql.shuffle.partitions = 2
SELECT * FROM df DISTRIBUTE BY KEY

那么，看看它是如何工作的：

par1: [(1,c), (3,b)]
par2: [(3,c), (1,b), (3,d)]
par3: [(3,a),(2,a)]

这将变成：

par1: [(1,c), (3,b), (3,c), (1,b), (3,d), (3,a)]
par2: [(2,a)]

排序方式

SELECT * FROM df SORT BY key

对于这种情况，它看起来像：

par1: [(1,c),  (1,b), (3,b), (3,c), (3,d), (3,a)]
par2: [(2,a)]

聚类依据

这是在同一组表达式上同时使用分发依据和排序依据的快捷方式。

SET spark.sql.shuffle.partitions =2
SELECT * FROM df CLUSTER BY key

注意：这是基本信息，如果有帮助，请告诉我，否则我们可以根据情况和设置使用各种不同的方法来优化您的 Spark 作业和查询。

【讨论】：