如何按键对Fs2流进行分区，分别对每个分区进行转换？

Question

我想要实现的，例如，，给定的数据：

time, part, data
0, a, 3
1, a, 4
2, b, 10
3, b, 20
3, a, 5

和转型：

stream.keyBy(_.part).scan(0)((s, d) => s + d)

得到：

0, a, 3
1, a, 7
2, b, 10
3, b, 30
3, a, 12

我尝试过使用groupAdjacentBy对其进行分区，但是它变得太复杂了，因为我需要用键在每个块之间保留复杂的状态。我想知道是否有类似于Flink DataStream.keyBy的东西？或者更简单的方法来实现它？

Answer 1

好的，我发现了有趣的解决方案 (虽然不能是flatten )

Answer 2

如前所述，这个问题可以通过扫描操作本身的“分区”来解决：

import cats.implicits._
import cats.effect.IO
import fs2._

case class Element(time: Long, part: Symbol, value: Int)

val elements = Stream(
  Element(0, 'a, 3),
  Element(1, 'a, 4),
  Element(2, 'b, 10),
  Element(3, 'b, 20),
  Element(3, 'a, 5)
)

val runningSumsByPart = elements
  .scan(Map.empty[Symbol, Int] -> none[Element]) {
    case ((sums, _), el@Element(_, part, value)) =>
      val sum = sums.getOrElse(part, 0) + value
      (sums + (part -> sum), el.copy(value = sum).some)
  }
  .collect { case (_, Some(el)) => el }

runningSumsByPart.covary[IO].evalTap(el => IO { println(el) }).compile.drain.unsafeRunSync()

产出：

元素(0，'a,3) 元素(1，'a,7) 元素(2，'b,10) 元素(3，'b,30) 元素(3，'a,12)

Answer 3

我做了这样的事。先分开，然后合并。我还不知道如何返回2流。我只知道如何在一个地方处理它们，然后将它们合并在一起。

    val notEqualS = in
      .filter(_.isInstanceOf[NotEqual])
      .map(_.asInstanceOf[NotEqual])
      ...

    val invalidS = in
      .filter(_.isInstanceOf[Invalid])
      .map(_.asInstanceOf[Invalid])
      ...

    notEqualS.merge(invalidS)