问题描述
我们有一个非常大的Pyspark数据框，需要在其上执行groupBy操作。
我们尝试过
df_gp=df.groupBy('some_column').count()
花费了很长时间（它已经运行了17多个小时，没有结果）。
我也尝试过
df_gp=df.groupBy('some_column').agg(count)
但据我所知，行为是相同的。
更多背景信息：
- 我们正在使用％spark2.pyspark解释器在Zeppelin（版本0.8.0）上运行此操作
- Zeppelin在纱线客户端上运行
- 数据存储在Hive（Hive 3.1.0.3.1.0.0-78）
- 通过使用llap查询Hive创建初始数据框：
from pyspark_llap import HiveWarehouseSession
hive = HiveWarehouseSession.session(spark).build()
req=""" SELECT *
FROM table
where isodate='2020-07-27'
"""
df = hive.executeQuery(req)
- 数据框大小约为6000万行，9列
- 在相同环境下对同一数据框执行的其他操作，例如
count()或cache()只需不到一分钟即可完成
我一直在不同来源上了解Spark的groupBy，但是从我收集的here来看，Dataframe API不需要加载或随机整理内存中的键，因此它不应该是即使在大型数据框上也存在问题。
我发现关于如此大量数据的groupBy可能会花费一些时间，但这确实太多了。我猜有些内存参数可能需要调整，或者我们执行groupBy操作的方式可能有问题吗？
[EDIT]我忘了提到groupBy之前在数据帧上正在处理一些UDF。我已经尝试过了：
-
groupBy在没有UDF的大型数据框中：不到一分钟即可得到结果 -
groupBy上已处理的数据帧的示例：与以前相同的问题
所以我们认为UDF是问题的实际原因，而不是groupBy
解决方法
首先是一些神话般的爆发
-
.groupBy('some_column').count()和.groupBy('some_column').count()相同 -
groupBy导致随机播放，该帖子的意思是，它仅仅随机播放必要的列数据（没有在groupBy或agg函数中未使用的多余列）我一直在不同的来源上了解Spark的groupBy，但是从我这里收集的数据来看，Dataframe API不需要加载或改组内存中的键，因此即使在大型Dataframe上也不应该成为问题。
现在遇到了问题
-
如果重新整理更多数据并且将
-
spark.sql.shuffle.partitions可能会花费一些时间。在这种情况下，1个核心将有大量的混洗数据进行聚合 - 如果
groupBy中使用的列存在数据偏斜，也可能会花费很多时间，因为它将导致大量数据流到单个执行器核心。
groupBy设置为低（默认值为200），解决方案
- 将
spark.sql.shuffle.partitions增加到一个更高的值（根据我的经验，应该确保<amount_of_data_shuffled_in_gb>/100MB左右，以确保1个内核获得大约100 MB的数据进行聚合 - 偏斜可以通过在数据中引入随机性（咸化）https://dzone.com/articles/why-your-spark-apps-are-slow-or-failing-part-ii-da 来解决。
运行速度很慢可能是由于基础Hive查询而不是由于groupBy操作引起的。您可能知道，spark会进行延迟评估，因此延迟可能来自上述两种情况。
一种测试方法是在执行groupBy之前cache()数据帧或调用简单的count()。如果您看到相同的问题，那是由于配置单元查询执行，解决方案在那里看起来会有所不同。您也可以尝试从文件中读取数据，看看执行groupBy时是否注意到相同的执行时间。