使用 Akka Streams 读取大文件答案

【问题标题】：Reading a large file using Akka Streams使用 Akka Streams 读取大文件
【发布时间】：2016-02-05 21:52:44
【问题描述】：

我正在试用 Akka Streams，这里有一个简短的 sn-p：

  override def main(args: Array[String]) {
    val filePath = "/Users/joe/Softwares/data/FoodFacts.csv"//args(0)

    val file = new File(filePath)
    println(file.getAbsolutePath)
    // read 1MB of file as a stream
    val fileSource = SynchronousFileSource(file, 1 * 1024 * 1024)
    val shaFlow = fileSource.map(chunk => {
      println(s"the string obtained is ${chunk.toString}")
    })
    shaFlow.to(Sink.foreach(println(_))).run // fails with a null pointer

    def sha256(s: String) = {
      val  messageDigest = MessageDigest.getInstance("SHA-256")
      messageDigest.digest(s.getBytes("UTF-8"))
    }
  }

当我运行这个 sn-p 时，我得到：

Exception in thread "main" java.lang.NullPointerException
    at akka.stream.scaladsl.RunnableGraph.run(Flow.scala:365)
    at com.test.api.consumer.DataScienceBoot$.main(DataScienceBoot.scala:30)
    at com.test.api.consumer.DataScienceBoot.main(DataScienceBoot.scala)

在我看来，不是 fileSource 只是空的吗？为什么是这样？有任何想法吗？ FoodFacts.csv 大小为 40MB，而我要做的就是创建一个 1MB 的数据流！

即使使用 8192 的 defaultChunkSize 也不起作用！

【问题讨论】：

您使用的是什么版本的 akka 流？我认为 SynchronousFileSource 现在已弃用
我用的是 1.0.我应该使用哪一个来读取一个大文件并将这些块作为流传递？有什么线索吗？

标签： scala akka-stream

【解决方案1】：

1.0 已被弃用。如果可以，请使用2.x。

当我尝试使用2.0.1 版本时，使用FileIO.fromFile(file) 而不是SynchronousFileSource，编译失败并显示消息fails with null pointer。这仅仅是因为它在范围内没有ActorMaterializer。包括它，让它工作：

object TImpl extends App {
import java.io.File

  implicit val system = ActorSystem("Sys")
  implicit val materializer = ActorMaterializer()

  val file = new File("somefile.csv")
  val fileSource = FileIO.fromFile(file,1 * 1024 * 1024 )
  val shaFlow: Source[String, Future[Long]] = fileSource.map(chunk => {
    s"the string obtained is ${chunk.toString()}"
  })

  shaFlow.runForeach(println(_))    
}

这适用于任何大小的文件。更多dispatcher配置请参考here。

【讨论】：