scala - 使用 Akka Streams 读取 CSV 文件

Question

我正在阅读一个 csv 文件。我正在使用 Akka Streams 来执行此操作，以便我可以创建要在每一行上执行的操作图。我已经启动并运行了以下玩具示例。

  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem("MyAkkaSystem")
    implicit val materializer = ActorMaterializer()

        val source = akka.stream.scaladsl.Source.fromIterator(Source.fromFile("a.csv").getLines)
        val sink = Sink.foreach(println)
        source.runWith(sink)
      }

两人 Source类型对我来说并不容易。这是惯用的还是有更好的写法？

Answer 1

其实，akka-streams提供直接读取文件的函数。

FileIO.fromPath(Paths.get("a.csv"))
      .via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8String))
      .runForeach(println)

在这里， runForeach方法是打印行。如果您有合适的 Sink要处理这些行，请使用它而不是此函数。例如，如果您想按 ' 分割线并打印其中的总字数:

val sink: Sink[String] = Sink.foreach(x => println(x.split(",").size))

FileIO.fromPath(Paths.get("a.csv"))
      .via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8String))
      .to(sink)
      .run()