纯Scala如何解析大体积CSV文件并按RECORDID计算指定列平均值

实现思路 #

核心采用逐行流式读取+局部状态匹配的方案，全程不加载全量文件数据，内存占用仅和未匹配到End的会话数量、去重后的RECORDID数量成正比，完全适配8GB/数百万行的文件处理需求：

1. 用哈希表暂存还未匹配到End状态的Begin会话记录
2. 每读到一条End状态的记录时，查询对应会话的Begin权重，计算差值后更新对应RECORDID的统计结果
3. 全部行处理完成后，遍历统计结果计算平均值输出即可

实现代码 #

import java.io.{BufferedReader, FileInputStream, InputStreamReader}
import scala.util.Using
import scala.collection.mutable

object SessionAvgCalc {
  def main(args: Array[String]): Unit = {
    // 存储未匹配End的会话：key为(RECORDID, SESSIONID)，value为Begin对应的WEIGHT
    val pendingSessionMap = mutable.HashMap[(Int, String), Int]()
    // 存储各RECORDID的统计结果：key为RECORDID，value为(差值总和, 有效会话数)
    val recordStatMap = mutable.HashMap[Int, (Double, Int)]()
    val targetFilePath = "/Users/Desktop/sessionfile.txt"

    // 自动管理IO流资源，避免泄漏
    Using(new BufferedReader(new InputStreamReader(new FileInputStream(targetFilePath)))) { reader =>
      Iterator.continually(reader.readLine())
        .takeWhile(_ != null)
        .filter(_.trim.nonEmpty) // 过滤空行
        .foreach { line =>
          val fields = line.split(",").map(_.trim)
          if (fields.length == 4) { // 跳过字段数量不符的脏数据
            try {
              val recordId = fields(0).toInt
              val weight = fields(1).toInt
              val status = fields(2)
              val sessionId = fields(3)

              status match {
                case "Begin" => pendingSessionMap.put((recordId, sessionId), weight)
                case "End" =>
                  // 匹配到对应Begin记录才计算差值
                  pendingSessionMap.remove((recordId, sessionId)).foreach { beginWeight =>
                    val diff = weight - beginWeight
                    // 更新统计数据
                    recordStatMap.updateWith(recordId) {
                      case Some((sum, cnt)) => Some((sum + diff, cnt + 1))
                      case None => Some((diff.toDouble, 1))
                    }
                  }
                case _ => // 其他状态直接忽略
              }
            } catch {
              case _: NumberFormatException => // 字段格式错误直接跳过
            }
          }
        }
    }.get

    // 按RECORDID升序输出结果
    recordStatMap.toSeq.sortBy(_._1).foreach { case (rid, (sum, cnt)) =>
      val avg = sum / cnt
      println(f"$rid, $avg%.1f")
    }
  }
}

方案优势 #

• 内存占用极低：仅存储未匹配的会话和统计结果，无全量数据加载，普通消费级机器即可处理
• 性能优异：整体时间复杂度为O(N)，哈希表操作均为O(1)，数百万行数据处理耗时通常在秒级
• 兼容性强：内置空行、脏数据、格式错误的异常处理逻辑，不会因为个别坏行导致程序崩溃
• 资源安全：通过Using语法自动释放IO资源，避免资源泄漏

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