如何在Scala中编写进程文件并并行编写结果？答案

【问题标题】：How to write process files and write the results in parallel in Scala?如何在Scala中编写进程文件并并行编写结果？
【发布时间】：2012-07-21 05:27:08
【问题描述】：

这是我的previous 问题的后续。

假设我并行处理我的文件。现在我想将处理结果写入文件。由于结果不适合内存，我不能等到所有文件的处理完成后再写入结果。我必须以某种方式并行进行处理和写入。

例如：假设我有带有数字的文件。文件大小约为500M。文件数约为200。每个文件都适合内存，但所有文件都不适合。现在我想将在这些文件中找到的所有偶数数写入另一个文件。

如何在 Scala 中做到这一点（使用 Futures 和 Scala parallel collections）？

【问题讨论】：

scalax.io 中的 Lines() 是懒惰评估的也看看future-exec jesseeichar.github.com/scala-io-doc/0.4.0/index.html#!/core/…

标签： file scala parallel-processing

【解决方案1】：

在某些时候你必须同步写作。如果您不想阻塞其他线程，一种可能性是使用参与者将结果写入文件。这可能如下所示：

class FileWriterActor(path: String) extends Actor {

  val file = ... // init FileWriter

  // this is how you implement an akka actor
  // plain scala actors look a bit different        
  def receive = {
    case x: MyResult => file.write(x.toString)
  }

  override def postStop() = file.close()
}

// usage
val result = ... // calculation stuff
fileWriter ! result

【讨论】：

这是一个很好的策略。这里可能不完全清楚的是线程在哪里。首先，actor 代表一个活动线程：它不断地在其输入队列中接收消息，并将它们按顺序写入文件。其次，所有客户端线程都使用“fileWriter！value”序列向（单个）actor 发送消息。这 '！'运算符是从 Hoare 的 CSP 代数中借来的，这个概念也是如此。作为替代方案，可以直接使用 CSP，例如通过 JCSP，以这种方式，这个演员的工作方式会更加明确。
在我的一个应用程序中，使用并行集合对大型输入文件的每一行执行 NLP，我从同步 println 切换到使用 Actor 来写入数据。 CPU 使用率从 180% 提高到 700%，每 100k 行的时间从 12 秒提高到 2.5 秒。

【解决方案2】：

对于不熟悉akka的人：

import java.io.{File, PrintWriter}
import akka.actor.{Actor,ActorSystem,Props}

object AkkaWriterExample extends App{

  val outputPath : String = ???
  val system = ActorSystem("WriterSystem")
  val writer = system.actorOf(Props(new WriterActor(new File(outputPath))), name="writer")
  writer ! "this is a test"
  system.shutdown()
  system.awaitTermination()
}

class WriterActor(outFile: File) extends Actor {

  val writer = new PrintWriter(outFile)

  // this is how you implement an akka actor
  // plain scala actors look a bit different        
  def receive = {
    case str:String => println(str); writer.write(str);
  }

  override def postStop() = {
    writer.flush(); 
    writer.close();
  }
}

【讨论】：