Spark Structured Streaming - groupByKey 按分区单独

Posted 2023-04-15

【中文标题】Spark Structured Streaming - groupByKey 按分区单独

【英文标题】：Spark Structured Streaming - groupByKey individually by partition

【发布时间】：2018-01-23 12:18:29

【问题描述】：

我的 Kafka 生产者正在根据给定的键将消息分发到主题分区中。

所以，在 Spark 端，我已经有需要在同一个分区中一起处理的消息。

现在，我需要做一个 groupByKey 以在每个分区中通过键在列表中聚合值，但不需要合并分区，因为没有机会在多个分区中拥有给定的键。

我怎样才能仅在分区级别执行此 groupByKey ？

|topic-partition1| ---> |spark-partition1| -- groupByKey --> |spark-partition1.1| -- mapGroupsWithState --> ...
|topic-partition2| ---> |spark-partition2| -- groupByKey --> |spark-partition2.1| -- mapGroupsWithState --> ...
|topic-partition3| ---> |spark-partition3| -- groupByKey --> |spark-partition3.1| -- mapGroupsWithState --> ...

【参考方案1】：

如果您知道所有事件都将进入给定的分区，您可以在数据集上使用DataSet.mapPartitions：

val dataSet: DataSet[(String, String)] = ???
dataSet.mapPartitions  iter =>
  val res: Map[String, List[(String, String)] =
    iter.toList.groupBy  case (key, _) => key 

  // Do additional processing on res, which is now grouped by each key
  // present in the partition.

否则，如果您需要mapGroupsWithState，则可以避免使用groupByKey，因为您需要KeyValueGroupedDataset[K, V]。

如果您关心性能，请不要担心，除非您在分析时发现这是一个瓶颈。

【讨论】：

是的@Yuval，它们在同一个分区中，但我需要使用 mapGroupsWithState 函数，我没有找到不调用 groupByKey 的方法来调用该函数。

@KleysonRios 没错，mapGroupsWithState 是在键值数据集上定义的。你不能没有它。

这意味着我没有办法实现这一点？使用 groupByKey.mapGroupsWithState，即使是洗牌，我也没有得到包含数据的多个分区。您可以查看我在此topic 上您之前帮助过我的后续回答吗？

@KleysonRios 你不能使用mapGroupsWithState，除非先按键分组。我不确定您所说的“我没有获得包含数据的多个分区”是什么意思，您能详细说明一下吗？

在 groupByKey.mapGroupsWithState 之前我有两个分区，后来我有两个分区（我猜）但所有数据都放在同一个分区中。 In this thread 我放了一些来自 Spark UI 的图像。而不是让两个任务并行运行，我只有一个。我读过一些关于分区算法的文章，但没有找到太多关于这个主题的内容。