如何在Spark Streaming中使用CombineByKey仅在分区内进行'reduce'？

Question

我已经通过Kafka将数据按键排序到我的Spark流分区中，即在一个节点上找到的键在任何其他节点上都找不到。

我希望使用redis及其incrby(increment by）命令作为状态引擎，并且为了减少发送到redis的请求数，我希望通过在每个工作节点上进行字数计数来部分减少数据。（关键是标签+时间戳，从字数中获取我的功能）。我希望避免洗牌，让redis负责跨工作节点添加数据。

即使我检查了数据在工作节点之间是否被清晰地分割，.reduce(_+_)(Scala语法）也会花费很长的时间（几秒，而对于map任务则是次秒），因为HashPartitioner似乎会将我的数据洗牌到一个随机节点以将其添加到那里。

如何在Spark Streaming中不触发Scala中的shuffling步骤的情况下，在每个分区器上编写一个简单的字数reduce？

注DStream对象缺少一些RDD方法，这些方法只有通过transform方法才可用。

看来我可以使用combinebykey。我想跳过mergeCombiners()步骤，而是将累积的元组留在原处。《学习的火花》一书谜一般地说：

如果我们知道数据不会从中受益，我们可以禁用combineByKey（）中的映射端聚合。例如，groupByKey()禁用映射端聚合，因为聚合函数（附加到列表）不会节省任何空间。如果我们想禁用映射端组合，我们需要指定分区器；现在，您可以通过传递RDD.partitioner在源RDD上使用分区器。

更糟糕的是，据我所知，在Spark Streaming中没有为DStream RDD设置分区器，所以我不知道如何为combineByKey提供一个分区器，它不会最终对数据进行洗牌。

此外，“地图端”实际上是什么意思，mapsidecombine=false会产生什么后果？

CombineByKey的scala实现可以在https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/pairrddfunctions上找到。scala查找CombineByKeyWithClasstag。

如果解决方案涉及自定义分区器，请同时包含一个代码示例，说明如何将该分区器应用于传入的DStream。

Answer 1

这可以使用mappartitions来完成，该函数将一个分区上的输入RDD的迭代器映射到输出RDD上的迭代器。

要实现字数计数，我映射到_.2以删除Kafka键，然后使用foldleft执行快速迭代器字数计数，初始化mutable.hashmap，然后将其转换为迭代器以形成输出RDD。

val myDstream = messages
  .mapPartitions( it =>
    it.map(_._2)
    .foldLeft(new mutable.HashMap[String, Int])(
      (count, key) => count += (key -> (count.getOrElse(key, 0) + 1))
    ).toIterator
  )