使用 fs2 处理带有内部流的流

Question

我有 2 个带有排序数据的 csv 文件： 文件 1：已排序的数字 (~1GB) 文件 2：排序后的数字 + 额外数据 (~20GB)

我需要在文件 2 中查找文件 1 中的所有数字并进行一些处理（文件 2 中不存在于文件 1 中的数字被跳过）。

到目前为止，我有：

object MainQueue extends IOApp {

  override def run(args: List[String]): IO[ExitCode] =
    program[IO].compile.drain.as(ExitCode.Success)

  def program[F[_]: Sync: ContextShift](): Stream[F, Unit] =
    for {
      number <- numberStream
      record <- records
                       .through(parser())
                       .through(findRecord(number))
      _ <- Stream.emit(println(s"$number <-> $record"))
    } yield ()

  def findRecord[F[_]](phone: Long): Pipe[F, Long, Long] =
    _.dropWhile(r => {
      println(s"Reading $r")
      r < phone
    }).head //halts the stream

  def numberStream[F[_]](): Stream[F, Long] =
    Stream(100L, 120L)

  //TODO: make stream continue and not halt and restart
  def records[F[_]: Sync: ContextShift](): Stream[F, String] =
    Stream
      .resource(Blocker[F])
      .flatMap { bec =>
        readAll[F](Paths.get("small.csv"), bec, 4096)
      }
      .through(text.utf8Decode)
      .through(text.lines)

  def parser[F[_]](): Pipe[F, String, Long] = ??? //parse

  def writer[F[_]](): Pipe[F, Long, Unit] =
    _.map(v => {
      println(s"Found: $v")
    })

}

哪些打印：

Reading 50
Reading 100
100 <-> 100
Reading 50
Reading 100
Reading 120
120 <-> 120

这意味着第二个流为文件 1 中的每个值重新启动，我如何保持上次读取的位置并从那里开始？数字是排序的，所以没有点重新开始。 我对 scala 和 fs2 非常陌生，因此非常感谢您解释我的误解。

谢谢！

Answer 1

首先你需要知道，那个

    for {
      number <- numberStream
      record <- records
                       .through(parser())
                       .through(findRecord(number))
      _ <- Stream.emit(println(s"$number <-> $record"))
    } yield ()

只是一个语法糖

    numberStream()
      .flatMap(number => records
        .through(parser())
        .through(findRecord(number)).map(x => (x, number)))
      .flatMap { case (record, number) => Stream.emit(println(s"$number <-> $record")) }

这意味着你评估一个效果，在这种情况下

records
        .through(parser())
        .through(findRecord(number)).map(x => (x, number))

，在numberStream的每个元素上。

要保留来自 File2 的指针的最后位置，您可以计算消耗的字节数，但您仍需要为 File1 中的每个数字重新打开与 File2 的连接。

您要实现的操作是条件压缩，因此您应该看看fs2.Stream#zip、fs2.Stream#zipAll 等方法，它可以帮助您只压缩这些打开文件的记录一次。

---

类似 Zip 的方法并不完全符合您的要求，但使用 fs2.Pull 实现请求的功能非常容易，这里有一个示例：

 def zipToLeft[F[_] : RaiseThrowable, O1, O2](in1: Stream[F, O1], in2: Stream[F, O2])
                             (condition: (O1, O2) => Boolean): Stream[F, (O1, O2)] = {
    def go(s1: Stream[F, O1], s2: Stream[F, O2]): Pull[F, (O1, O2), Unit] =
      s1.pull.uncons1.flatMap {
        case Some((hd1, tl1)) => s2.pull.uncons1.flatMap {
          case Some((hd2, tl2)) => if (condition(hd1, hd2)) Pull.output1((hd1, hd2)) >> go(tl1, tl2)
              else go(s1, tl2)
          case None => Pull.raiseError[F](new RuntimeException)
        }
        case None =>Pull.done
      }

    go(in1, in2).stream
}

你可以这样使用它：

    result <- program[IO].compile.toList
    _ <- IO(println(result))
  } yield ExitCode.Success

  def program[F[_] : Sync : ContextShift]() = zipToLeft(numberStream(), records()) { case (v1, v2) => v1 == v2._1 }

  def numberStream[F[_]](): Stream[F, Long] =
    Stream.emits(Vector(1, 3, 6, 7, 9))

  def records[F[_] : Sync : ContextShift](): Stream[F, (Int, String)] =
    Stream.emits(Vector.range(1, 10).map(i => (i, i.toString)))

输出：List((1,(1,1)), (3,(3,3)), (6,(6,6)), (7,(7,7)), (9,(9,9)))