Akka流:使用标题处理CSV

Question

使用Akka Streams读取CSV文件 - 基于此问题.我使用Akka Streams读取CSV.现在我需要逐行执行它 - 但我还需要知道标题的名称是什么.任何选择？

UPD.澄清一点.

FileIO.fromPath(Paths.get("a.csv))
.via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8String))
.runForeach(println /* header  + current line for each line*/)

Answer 1

您可以使用prefixAndTail第一个元素 - 在本例中为标题 - 然后将其与后续元素(行)合并.

这是一个示例,标题和列组合在一起形成Map[String, String]:

val flow: Flow[Seq[String], Map[String, String], NotUsed] = Flow[Seq[String]]
  .prefixAndTail(1).flatMapConcat { case (headers, rows) =>
    rows.map (row => headers.head.zip(row).toMap)
  }

val test: Source[Seq[String], NotUsed] = Source(
  List(Seq("col1", "col2"), Seq("a", "b"), Seq("1", "2")))

Await.result(test.via(flow).runForeach(println), 20.seconds)
// Map(col1 -> a, col2 -> b)
// Map(col1 -> 1, col2 -> 2)