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

Question

我有 2 个包含排序数据的 csv 文件:文件 1:排序的数字 (~1GB)文件 2:排序的数字 + 额外数据(~20GB)

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

到目前为止我有:

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 的 super 新手，所以非常感谢对我误解的解释。

谢谢!

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)))