使用 mapPartitions 避免使用 groupby 和 count 进行洗牌答案

【问题标题】：Using mapPartitions to avoid the shuffle with groupby and count使用 mapPartitions 避免使用 groupby 和 count 进行洗牌
【发布时间】：2019-10-01 07:48:23
【问题描述】：

我有以下数据，我需要根据键分组并根据键计数以监控指标。我可以使用 groupBy 并为该组进行计数，但这涉及一些洗牌。我们可以不做 shuffle 吗？

ID,TempID,PermanantID
----------

xxx, abcd, 12345

xxx, efg, 1345

xxx, ijk, 1534

xxx, lmn, 13455

xxx, null, 12345

xxx, axg, null

yyy, abcd, 12345

yyy, efg, 1345

yyy, ijk, 1534

zzz, lmn, 13455

zzz, abc, null

输出应该是

ID Count1 Count2
----------
XXX 5 5

YYY 3 3

ZZZ 2 1

我可以用 groupBy 和 count 做到这一点

dataframe.groupby("ID").agg(col("TempID").as("Count1"),count(col("PermanantID").as("Count2"))

我们可以使用 mapPartition 做到这一点吗？

【问题讨论】：

标签： scala dataframe apache-spark streaming spark-structured-streaming

【解决方案1】：

这个问题虽然可以理解，但有缺陷。

mapPartitions 不能直接用于数据帧，而是用于 RDD 和数据集。

此外，在调用 mapPartitions 之前所需的分区和洗牌又如何呢？否则，结果将不正确。问题中最初没有提到数据顺序的保证。

因此，我将依赖假设的 groupBy 方法。认为 App 中不需要 shuffle 是一种错觉，而是我们可以减少 shuffle，这就是我们的目标。

【讨论】：

我想获取每个分区的这些统计信息并发送到我可以选择聚合的外部系统，是的，目标是减少随机播放，但从地图分区上的信息显示，将避免随机播放，因为您在具有一个或多个分区的每个执行程序上进行计算或聚合。它的火花结构化流媒体..所以期望数据来自kafka，带有微批量..我可以看到dataframe.mappartion函数。为什么我们不能在数据帧上做？
这些是额外的问题。发布新问题。 Spark 不适合带有 mapP 的 df。顺便说一句，我知道它是如何工作的。
好的。我的问题保持不变。如果我将其转换为数据集，我将能够进行 mappartition。但是可以对 mappartion 提供的迭代器进行分组和聚合吗？
是的，但您必须先正确分区。所有的 xxxx 都需要在同一个分区中。
所以你无法避免洗牌。