提高涉及文件转换的 fs2 流的性能

Question

我有这样的东西（这是来自 https://github.com/typelevel/fs2 的例子，加上我的补充，我用评论标记）：

import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource}
import fs2.{io, text, Stream}
import java.nio.file.Paths

object Converter extends IOApp {

  val converter: Stream[IO, Unit] = Stream.resource(Blocker[IO]).flatMap  { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
      .balanceAvailable // my addition
      .map ( worker => // my addition
        worker // my addition
          .through(text.utf8Decode)
          .through(text.lines)
          .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
          .map(line => fahrenheitToCelsius(line.toDouble).toString)
          .intersperse("\n")
          .through(text.utf8Encode)
          .through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
      ) // my addition
      .take(4).parJoinUnbounded // my addition
  }

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

如果fahrenheit.txt和eg一样大。 300mb 原始代码的执行需要几分钟。看来我的代码并没有更快。我怎样才能提高它的性能？运行的时候还有很多unusedCPU电量，盘是SSD的，不知道为什么这么慢。我不确定我是否正确使用 balance。

Answer 1

罪魁祸首是 text.utf8Encode，它不必要地每行发出一个块。当有很多短线时，这是一个巨大的开销，就像在示例中一样（每行一个温度值，108199750 行）。最近解决了（Pull request：https://github.com/typelevel/fs2/pull/2096）。下面我提供了一个基于此 PR 的内联解决方案（只要有人使用没有此修复的版本就很有用）：

import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource}
import fs2.{io, text, Stream, Pipe, Chunk}
import java.nio.file.Paths
import java.nio.charset.Charset

object Converter extends IOApp {

  val converter: Stream[IO, Unit] = Stream.resource(Blocker[IO]).flatMap  { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    def betterUtf8Encode[F[_]]: Pipe[F, String, Byte] =
      _.mapChunks(c => c.flatMap(s => Chunk.bytes(s.getBytes(Charset.forName("UTF-8")))))

    io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
      .through(text.utf8Decode)
      .through(text.lines)
      .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
      .map(line => fahrenheitToCelsius(line.toDouble).toString)
      .intersperse("\n")
      // .through(text.utf8Encode) // didn't finish, could be an hour
      .through(betterUtf8Encode) // 2 minutes
      .through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
  }

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

2 分钟和可能一个小时或更长时间之间的差异，在这种情况下...

提高涉及文件转换的 fs2 流的性能

Improving performance of fs2 stream involving file transformation

scala

fs2

cats-effect