Чтение файлов CSV с использованием Akka Streams - на основе этого вопроса. У меня есть чтение CSV с использованием Akka Streams. Теперь мне нужно обработать его построчно, но мне также нужно знать, как назывались заголовки. Есть варианты?
УПД. Немного уточняю.
FileIO.fromPath(Paths.get("a.csv))
.via(Framing.delimiter(ByteString("\n"), 256, true).map(_.utf8String))
.runForeach(println /* header + current line for each line*/)
Вы можете использовать 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)
Alpakka, набор коннекторов Akka Streams, обеспечивает поддержку CSV:
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)
