Pyspark组通过大数据框 首先是一些神话般的爆发现在遇到了问题解决方案

如何解决Pyspark组通过大数据框 首先是一些神话般的爆发现在遇到了问题解决方案

我们有一个非常大的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

解决方法

首先是一些神话般的爆发

1. .groupBy('some_column').count()和.groupBy('some_column').count()相同

2. groupBy导致随机播放，该帖子的意思是，它仅仅随机播放必要的列数据（没有在groupBy或agg函数中未使用的多余列）

   我一直在不同的来源上了解Spark的groupBy，但是从我这里收集的数据来看，Dataframe API不需要加载或改组内存中的键，因此即使在大型Dataframe上也不应该成为问题。

现在遇到了问题

如果重新整理更多数据并且将groupBy设置为低（默认值为200），
1. spark.sql.shuffle.partitions可能会花费一些时间。在这种情况下，1个核心将有大量的混洗数据进行聚合
2. 如果groupBy中使用的列存在数据偏斜，也可能会花费很多时间，因为它将导致大量数据流到单个执行器核心。

解决方案

1. 将spark.sql.shuffle.partitions增加到一个更高的值（根据我的经验，应该确保<amount_of_data_shuffled_in_gb>/100MB左右，以确保1个内核获得大约100 MB的数据进行聚合
2. 偏斜可以通过在数据中引入随机性（咸化）https://dzone.com/articles/why-your-spark-apps-are-slow-or-failing-part-ii-da
来解决。

,

运行速度很慢可能是由于基础Hive查询而不是由于groupBy操作引起的。您可能知道，spark会进行延迟评估，因此延迟可能来自上述两种情况。 一种测试方法是在执行groupBy之前cache()数据帧或调用简单的count()。如果您看到相同的问题，那是由于配置单元查询执行，解决方案在那里看起来会有所不同。您也可以尝试从文件中读取数据，看看执行groupBy时是否注意到相同的执行时间。