Akka&Alpakka技术问题：如何在流处理中跨图阶段传递文件名信息

要给每条CSV记录带上文件名信息，其实可以在流处理的每一步把文件名上下文和行数据结合起来，这里有几种简单直接的实现方式：

单个文件的处理 #

如果只是处理单个文件，你只需要在解析CSV行之后，把文件名作为额外字段添加到每一行的字段列表里就行：

import akka.stream.alpakka.csv.scaladsl.CsvParsing
import akka.stream.scaladsl.FileIO
import java.nio.file.Paths
import akka.util.ByteString

val filename = "10002070.csv"
val filenameByteString = ByteString(filename)

val source = FileIO.fromPath(Paths.get(filename))
  .via(CsvParsing.lineScanner())
  // 把文件名作为第一个字段添加到每条CSV记录中
  .map(csvLine => filenameByteString :: csvLine)

这样处理后，每一条输出的List[ByteString]都会以文件名开头，后面跟着原来的CSV字段。

多个文件合并处理 #

如果要处理多个文件并把它们接入同一条流，你可以先创建一个包含所有文件路径的Source，然后对每个文件路径单独处理，再把所有流合并起来：

import akka.stream.scaladsl.Source

val filePaths = List("10002070.csv", "sales-2024.csv", "user-data.csv")

val combinedSource = Source(filePaths)
  .flatMapConcat { filePath =>
    // 从路径中提取文件名（根据你的路径格式调整逻辑）
    val filename = Paths.get(filePath).getFileName.toString
    val filenameBs = ByteString(filename)
    
    FileIO.fromPath(Paths.get(filePath))
      .via(CsvParsing.lineScanner())
      .map(csvLine => filenameBs :: csvLine)
  }

这样每个文件的每一行记录都会带上自己的文件名，所有文件的流会被合并成一条连续的处理流。

处理表头（可选） #

如果你的CSV文件包含表头，想要给表头也添加对应的字段名（比如source_file），可以稍微调整map逻辑：

val filename = "10002070.csv"
val filenameBs = ByteString(filename)
val headerFieldName = ByteString("source_file")

val source = FileIO.fromPath(Paths.get(filename))
  .via(CsvParsing.lineScanner())
  .zipWithIndex
  .map { case (csvLine, index) =>
    if (index == 0) {
      // 第一行是表头，添加字段名
      headerFieldName :: csvLine
    } else {
      // 数据行添加文件名
      filenameBs :: csvLine
    }
  }

这里用zipWithIndex来区分表头行（索引0）和数据行，确保表头和数据的字段对应上。

这种方式的核心思路是把文件的元数据（文件名）和文件内容的流绑定在一起，在每一行数据处理时都带上这个元信息，完全符合Akka Streams的流式处理模型，不会因为文件大小影响性能。

内容的提问来源于stack exchange，提问作者Falk Schuetzenmeister