本文介绍了Akka 流:读取多个文件的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
将所有这些作为单一来源读取. 文件应该按顺序读取.(无循环) 任何时候都不应该要求任何文件完全在内存中. 从文件读取错误应该折叠流.
我有一个文件列表.我要:
I have a list of files. I want:
感觉这应该可行:(Scala,akka-streams v2.4.7)
It felt like this should work: (Scala, akka-streams v2.4.7)
val sources = Seq("file1", "file2").map(new File(_)).map(f => FileIO.fromPath(f.toPath) .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true)) .map(bs => bs.utf8String) ) val source = sources.reduce( (a, b) => Sourcebine(a, b)(MergePreferred(_)) ) source.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) // counting lines但这会导致编译错误,因为 FileIO 有一个与之关联的物化值,而 Sourcebine 不支持.
But that results in a compile error since FileIO has a materialized value associated with it, and Sourcebine doesn't support that.
映射物化值让我想知道文件读取错误是如何处理的,但确实可以编译:
Mapping the materialized value away makes me wonder how file-read errors get handled, but does compile:
val sources = Seq("file1", "file2").map(new File(_)).map(f => FileIO.fromPath(f.toPath) .via(Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true)) .map(bs => bs.utf8String) .mapMaterializedValue(f => NotUsed.getInstance()) ) val source = sources.reduce( (a, b) => Sourcebine(a, b)(MergePreferred(_)) ) source.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) // counting lines但在运行时抛出 IllegalArgumentException:
But throws an IllegalArgumentException at runtime:
java.lang.IllegalArgumentException: requirement failed: The inlets [] and outlets [MergePreferred.out] must correspond to the inlets [MergePreferred.preferred] and outlets [MergePreferred.out] 推荐答案为了清楚地模块化不同的关注点,下面的代码没有尽可能简洁.
The code below is not as terse as it could be, in order to clearly modularize the different concerns.
// Given a stream of bytestrings delimited by the system line separator we can get lines represented as Strings val lines = Framing.delimiter(ByteString(System.lineSeparator), 10000, allowTruncation = true).map(bs => bs.utf8String) // given as stream of Paths we read those files and count the number of lines val lineCounter = Flow[Path].flatMapConcat(path => FileIO.fromPath(path).via(lines)).fold(0l)((count, line) => count + 1).toMat(Sink.head)(Keep.right) // Here's our test data source (replace paths with real paths) val testFiles = Source(List("somePathToFile1", "somePathToFile2").map(new File(_).toPath)) // Runs the line counter over the test files, returns a Future, which contains the number of lines, which we then print out to the console when it completes testFiles runWith lineCounter foreach println更多推荐
Akka 流:读取多个文件
发布评论