如何使用同一流中存在的键对流的元素进行分组？答案

【问题标题】：How do you group the elements of a stream using keys present in the same stream?如何使用同一流中存在的键对流的元素进行分组？
【发布时间】：2019-10-27 07:26:48
【问题描述】：

所以我有以下类型：

case class Episode(
  parentTconst: String,
  seasonNumber: Int,
  episodeNumber: Int
)

以及以下来源：

val episodeSource: Source[Episode, _] = FileIO.fromPath(Paths.get(myDataFilePath)).via(myDataParserToEpisode)

剧集的示例列表是：

Seq(
  Episode("gameof", 5, 8),
  Episode("mentalist", 2, 4),
  Episode("gameof", 5, 8),
  Episode("mentalist", 1, 8),
  Episode("rikiandmanual", 1, 8)
)

我正在尝试制作以下流（我不确定它是否是正确的签名，请指教，也许它是一个接收器，因为我正在消耗整个流来获取地图）：

def gimmeThoseEpisodeGroups: Flow[Episode, Map[String, Seq[Episode]], _]

如果打印出来会是这样的：

Map(
  "gameof"        -> Seq(Episode("gameof", 5, 8), Episode("gameof", 5, 8)),
  "mentalist"     -> Seq(Episode("mentalist", 2, 4), Episode("mentalist", 1, 8)),
  "rikiandmanual" -> Seq(Episode("rikiandmanual", 1, 8))
)

我尝试了很多组合都不起作用。可能方法不对。

我想我应该使用 groupBy。另一件事是我可能需要第一次使用流来获取组的键列表（或者可能找到一种方法来使用它一次但产生两个结果）。

我 DuckDucked 并发现了有关广播的信息，但我还没有完全理解它。

    def getGroupedByParentTConst: Flow[Episode, Map[String, Seq[Episode]], _] = Flow[Episode].groupBy(2, _.parentTconst)

但这失败了（无论如何，我觉得这不是正确的道路。

另外，我想我会提取 parentTconst 以使用它们作为分组键：

    def getParentTConst: Flow[Episode, Set[String], _] = Flow[Episode].fold(Set.empty[String]) {
      (right, left) => { right ++ Set(left.parentTconst) }
    }

这行得通，但我一直在弄清楚如何使用它们来对我的原始来源进行分组......

另外，如果我使用键列表，这意味着我必须消耗两个流来进行分组。

我想我有一个 Duck 问题，这不是一个非常困难的问题，但是由于我有使用 akka-streams 的限制，所以它并不是很简单。

任何帮助将不胜感激。

【问题讨论】：

预期类型错误。 Flow[Episode, Map[String, Seq[Episode]], _] 表示每个Episode 都将转换为Map[String, Seq[Episode]]，而不是批量和聚合。

标签： scala akka akka-stream

【解决方案1】：

您试图实现的目标无法以流媒体方式完成，因为创建所有分组剧集的地图需要读取内存中的所有数据。没有任何中间结果。

如果您仍然想这样做，您可以按照与您提出的方式类似的方式使用 fold

Flow[Episode].fold(Map.empty[String, List[Episode]]) { (map, e) ⇒
  val key = e.parentTconst
  map + (key → v :: map.getOrElse(key, Nil))
}

但由于这将读取内存中的所有文件，您不妨省去使用 akka-streams 的麻烦，只需使用scala.io.Source。

如果您有太多数据无法放入内存，那么您需要更改要求。

【讨论】：