使用 Akka Streams 读取大文件
Reading a large file using Akka Streams
我正在试用 Akka Streams,这是我拥有的一个简短片段:
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"))
}
}
当我 运行 这个片段时,我得到:
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 也不起作用!
好吧 1.0
已弃用。如果可以,请使用 2.x
.
当我尝试使用 FileIO.fromFile(file)
而不是 SynchronousFileSource
来使用 2.0.1
版本时,编译失败并显示消息 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(_))
}
这适用于任何大小的文件。有关调度程序配置的更多信息,请参阅 here.
我正在试用 Akka Streams,这是我拥有的一个简短片段:
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"))
}
}
当我 运行 这个片段时,我得到:
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 也不起作用!
好吧 1.0
已弃用。如果可以,请使用 2.x
.
当我尝试使用 FileIO.fromFile(file)
而不是 SynchronousFileSource
来使用 2.0.1
版本时,编译失败并显示消息 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(_))
}
这适用于任何大小的文件。有关调度程序配置的更多信息,请参阅 here.