Akka：如何在一个图形阶段提取一个值并在下一个阶段使用它答案

【问题标题】：Akka: How to extract a value in one graph stage and use it in the nextAkka：如何在一个图形阶段提取一个值并在下一个阶段使用它
【发布时间】：2018-03-27 07:34:38
【问题描述】：

我正在使用 Alpakka 和 Akka 来处理 CSV 文件。由于我有一堆 CSV 文件必须添加到同一个流中，我想添加一个包含来自文件名或请求的信息的字段。目前我有这样的事情：

val source = FileIO.fromPath(Paths.get("10002070.csv"))
  .via(CsvParsing.lineScanner())

它流式传输字节字符串（字段）的列表（行）序列。目标是这样的：

val filename = "10002070.csv"
val source = FileIO.fromPath(Path.get(filename))
    .via(CsvParsing.lineScanner())
    .via(AddCSVFieldHere(filename))

创建类似于以下内容的结构：

10002070.csv,max,estimated,12,1,0

文件名是原始源中不存在的字段。

我认为在流中注入值看起来不是很漂亮，而且最终我想确定在读取目录的流阶段传递给解析的文件名。

通过流阶段传递值以供以后重用的正确/规范方法是什么？

【问题讨论】：

我会使用扇出，一个流用于内容，另一个用于名称文件，并在必要时压缩...doc.akka.io/docs/akka/2.5.3/scala/stream/stream-graphs.html
是的，这是我正在考虑的解决方案。我遇到的问题是，在扇出和处理后，流的元素数量与行数不同。当文件流结束（文件具有不同的大小）时，如何确保来自目录读取的流提前一次。也许 mergePreferred 会这样做，但我有点害怕竞争条件。
您是否要为整个目录创建一个流？也许最好为每个文件生成一个流。我认为这可能会更好，否则您应该创建某种加入流程..如果我理解得很好..

标签： scala akka akka-stream alpakka

【解决方案1】：

您可以使用map 转换流以将文件名添加到每个List[ByteString]：

val fileName = "10002070.csv"
val source =
  FileIO.fromPath(Path.get(fileName))
    .via(CsvParsing.lineScanner())
    .map(List(ByteString(fileName)) ++ _)

例如：

Source.single(ByteString("""header1,header2,header3
                           |1,2,3
                           |4,5,6""".stripMargin))
  .via(CsvParsing.lineScanner())
  .map(List(ByteString("myfile.csv")) ++ _)
  .runForeach(row => println(row.map(_.utf8String)))

// The above code prints the following:
// List(myfile.csv, header1, header2, header3)
// List(myfile.csv, 1, 2, 3)
// List(myfile.csv, 4, 5, 6)

同样的方法适用于您不知道文件名的更一般情况。如果您想读取目录中的所有文件（假设所有这些文件都是 csv 文件），将文件连接到单个流中，并在每个流元素中保留文件名，那么您可以使用 Alpakka 的 @987654321 @实用程序以下列方式：

val source =
  Directory.ls(Paths.get("/my/dir")) // Source[Path, NotUsed]
    .flatMapConcat { path =>
       FileIO.fromPath(path)
         .via(CsvParsing.lineScanner())
         .map(List(ByteString(path.getFileName.toString)) ++ _)
    }

【讨论】：