Geek Answers Handbook

Akka Streams: Reading Multiple Files

I have a list of files. I want to:

  • To read them all as one source.
  • Files should be read sequentially, in order. (no circular motion)
  • In no case should you require that any file be completely in memory.
  • An error reading from a file should minimize the stream.

It seemed 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) => Source.combine(a, b)(MergePreferred(_)) ) source.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) // counting lines

But this leads to a compilation error, since FileIO has a materialized value associated with it, and Source.combine does not support this.

Displaying a materialized value makes me wonder how file read errors are handled, but it compiles:

 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) => Source.combine(a, b)(MergePreferred(_)) ) source.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) // counting lines

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]
+5
source share
3 answers

The code below is not so brief so that it can clearly separate the various problems.

 // 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 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
+8
source

Refresh Oh, I did not see the accepted answer because I did not refresh the page> _ <. I will leave it here anyway, as I also added some notes about error handling.

I believe the following program does what you want:

 import akka.NotUsed import akka.actor.ActorSystem import akka.stream.{ActorMaterializer, IOResult} import akka.stream.scaladsl.{FileIO, Flow, Framing, Keep, Sink, Source} import akka.util.ByteString import scala.concurrent.{Await, Future} import scala.util.{Failure, Success} import scala.util.control.NonFatal import java.nio.file.Paths import scala.concurrent.duration._ object TestMain extends App { implicit val actorSystem = ActorSystem("test") implicit val materializer = ActorMaterializer() implicit def ec = actorSystem.dispatcher val sources = Vector("build.sbt", ".gitignore") .map(Paths.get(_)) .map(p => FileIO.fromPath(p) .viaMat(Framing.delimiter(ByteString(System.lineSeparator()), Int.MaxValue, allowTruncation = true))(Keep.left) .mapMaterializedValue { f => f.onComplete { case Success(r) if r.wasSuccessful => println(s"Read ${r.count} bytes from $p") case Success(r) => println(s"Something went wrong when reading $p: ${r.getError}") case Failure(NonFatal(e)) => println(s"Something went wrong when reading $p: $e") } NotUsed } ) val finalSource = Source(sources).flatMapConcat(identity) val result = finalSource.map(_ => 1).runWith(Sink.reduce[Int](_ + _)) result.onComplete { case Success(n) => println(s"Read $n lines total") case Failure(e) => println(s"Reading failed: $e") } Await.ready(result, 10.seconds) actorSystem.terminate() }

The key point is the flatMapConcat() method: it converts each element of the stream to a source and returns the stream of elements provided by these sources if they are executed sequentially.

As for error handling, you can add a handler to the future in the mapMaterializedValue argument mapMaterializedValue or you can handle the final error of a working thread by placing the handler in the materialized future value of Sink.foreach . I did both in the above example, and if you test it in, say, a non-existent file, you will see that the same error message will be printed twice. Unfortunately, flatMapConcat() does not collect materialized values, and to be honest, I don’t see how it could do it safely, so you have to handle them separately if necessary.

+2
source

I have one answer from the gate - do not use akka.FileIO . This seems to work fine, for example:

 val sources = Seq("sample.txt", "sample2.txt").map(io.Source.fromFile(_).getLines()).reduce(_ ++ _) val source = Source.fromIterator[String](() => sources) val lineCount = source.map(_ => 1).runWith(Sink.reduce[Int](_ + _))

I would still like to know if there is a better solution.

-1
source

More articles:


All Articles