性能:按以前分组列的子集进行分组

Question

我有一个带有两个分类列的DataFrame，类似于以下示例：

+----+-------+-------+
| ID | Cat A | Cat B |
+----+-------+-------+
|  1 |   A   |   B   |
|  2 |   B   |   C   |
|  5 |   A   |   B   |
|  7 |   B   |   C   |
|  8 |   A   |   C   |
+----+-------+-------+

我有一些处理需要两个步骤：第一个步骤需要将数据按两个分类列分组。在该示例中，它将生成以下DataFrame：

+-------+-------+-----+
| Cat A | Cat B | Cnt |
+-------+-------+-----+
|   A   |   B   |  2  |
|   B   |   C   |  2  |
|   A   |   C   |  1  |
+-------+-------+-----+

然后，下一步只按CatA进行分组，以计算新的聚合，例如：

+-----+-----+
| Cat | Cnt |
+-----+-----+
|  A  |  3  |
|  B  |  2  |
+-----+-----+

现在是问题：

1. 在我的解决方案中，我通过创建中间数据帧

   val df2 = df.groupBy("catA", "catB").agg(...)
   

   然后我把这个CCD_ 1聚合得到最后一个：

   val df3 = df2.groupBy("catA").agg(...)
   

   我认为它比再次聚集第一个DF更有效。这是一个好的假设吗？或者这没什么区别？

2. 有没有什么建议可以更有效地实现同样的结果？

Answer 1

一般来说，这看起来是一种很好的方法，应该比两次聚合数据更高效。由于shuffle文件是隐式缓存的，所以至少部分工作应该只执行一次。因此，当您在df2上调用操作并随后在df3上调用操作时，您应该会看到与df2对应的阶段已被跳过。此外，由第一混洗实施的部分结构可以减少在第二agg期间对聚合缓冲器的存储器需求。

遗憾的是，与RDD聚合不同，DataFrame聚合不能使用自定义分区器。这意味着您不能使用基于catA值的单个混洗来计算两个数据帧。这意味着第二次聚合将需要单独的交换散列分区。我怀疑它是否有理由切换到RDDs。