Akka 流：使用 header 处理 CSV

Question

Reading a CSV files using Akka Streams - 基于这个问题。我使用 Akka Streams 读取了 CSV。现在我需要逐行执行处理 - 但我还需要知道 headers 的名称是什么。有什么选择吗？

更新。澄清一点。

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 获取第一个元素 - 在本例中为 headers - 然后将其与后续元素（行）延迟组合。

下面是 headers 和列组合形成 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)

Answer 2

Alpakka，Akka Streams 连接器集合，提供CSV support:

Source
  .single(ByteString("""header1,header2,header3
                       |1,2,3
                       |4,5,6""".stripMargin))
  .via(CsvParsing.lineScanner())
  .via(CsvToMap.toMap())
  .map(_.mapValues(_.utf8String))
  .runForeach(println)

// Map(header1 -> 1, header2 -> 2, header3 -> 3)
// Map(header1 -> 4, header2 -> 5, header3 -> 6)

Akka 流：使用 header 处理 CSV

Akka stream: process CSV with header

scala

akka

akka-stream