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

谢谢！

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

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

Process Stream with inner stream with fs2

scala

fs2