使用 Akka Streams 读取 CSV 文件

问题描述

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

I'm reading a csv file. I am using Akka Streams to do this so that I can create a graph of actions to perform on each line. I've got the following toy example up and running.

  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 类型对我来说并不容易.这是惯用的还是有更好的写法?

The two Source types don't sit easy with me. Is this idiomatic or is there is a better way to write this?

推荐答案

其实，akka-streams 提供了直接读取文件的功能.

Actually, akka-streams provides a function to directly read from a file.

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

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

Here, runForeach method is to print the lines. If you have a proper Sink to process these lines, use it instead of this function. For example, if you want to split the lines by ' and print the total number of words in it:

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

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

这篇关于使用 Akka Streams 读取 CSV 文件的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！