我如何加快scalaz-stream文本处理

Question

如何加快以下scalaz-stream代码？目前，处理70MB的文本大约需要5分钟，所以我可能做错了什么，因为普通的Scala等效物需要几秒钟。

（跟进另一个问题）

  val converter2: Task[Unit] = {
    val docSep = "~~~"
    io.linesR("myInput.txt")
      .flatMap(line => { val words = line.split(" ");
          if (words.length==0 || words(0)!=docSep) Process(line)
          else Process(docSep, words.tail.mkString(" ")) })
      .split(_ == docSep)
      .filter(_ != Vector())
      .map(lines => lines.head + ": " + lines.tail.mkString(" "))
      .intersperse("n")
      .pipe(text.utf8Encode)
      .to(io.fileChunkW("correctButSlowOutput.txt"))
      .run
  }

Answer 1

我认为您可以将其中一种process1块方法用于块。如果您希望在线路的合并中进行很多并行处理，请确定有序输出是否重要，并使用合并或TEE的通道。这也将使它重复使用。因为您进行的处理很少，所以您可能会被开销淹没，因此您必须更加努力地使您的工作单位足够大，以免被淹没。

Answer 2

以下内容基于 @user1763729的块建议。虽然感觉很笨拙，而且就像原始版本一样慢。

  val converter: Task[Unit] = {
    val docSep = "~~~"
    io.linesR("myInput.txt")
      .intersperse("n") // handle empty documents (chunkBy has to switch from true to false)
      .zipWithPrevious // chunkBy cuts only *after* the predicate turns false
      .chunkBy{ 
        case (Some(prev), line) => { val words = line.split(" "); words.length == 0 || words(0) != docSep } 
        case (None, line) => true }
      .map(_.map(_._1.getOrElse(""))) // get previous element
      .map(_.filter(!Set("", "n").contains(_)))
      .map(lines => lines.head.split(" ").tail.mkString(" ") + ": " + lines.tail.mkString(" "))
      .intersperse("n")
      .pipe(text.utf8Encode)
      .to(io.fileChunkW("stillSlowOutput.txt"))
      .run
  }

编辑：

实际上，执行以下操作（只是阅读文件，没有写作或处理）已经需要1.5分钟，所以我想没有太多希望加快这一点的希望。

  val converter: Task[Unit] = {
    io.linesR("myInput.txt")
      .pipe(text.utf8Encode)
      .run
  }