如何基于数据标识/首尾标识过滤CSV的Spark RDD并处理Option[String]

核心问题 #

原代码直接调用dataIndicator.get会在dataIndicator为None时抛出NoSuchElementException，必须对Option[String]做安全分支处理，同时实现两种过滤逻辑。

代码实现 #

val rdd = spark.sparkContext.textFile(path)

val filteredRdd = dataIndicator match {
    // 当存在数据标识时，过滤以该标识开头的记录
    case Some(indicator) =>
        rdd.filter(_.startsWith(indicator))
    // 当数据标识缺失时，基于文件头尾标识过滤（示例为排除首尾行，可根据实际标识规则调整）
    case None =>
        // 安全获取文件首行（头部标识行）和末行（尾部标识行）
        val headerLine = rdd.first()
        val footerLine = rdd.reduce((_, last) => last)
        // 过滤掉头尾标识行，保留中间有效记录
        rdd.filter(line => line != headerLine && line != footerLine)
}

关键说明 #

1. Option安全处理：通过Scala模式匹配match分支处理Some/None两种情况，避免直接调用get引发异常。
2. 头尾标识过滤逻辑调整：示例中默认排除首尾行，若你的头尾标识是特定前缀（如HEADER_/FOOTER_），可修改过滤条件为：

   rdd.filter(line => !line.startsWith("HEADER_") && !line.startsWith("FOOTER_"))
   

3. 大文件适配：用first()和reduce((_, last) => last)获取首尾行，避免collect()加载全量数据到Driver内存引发OOM。

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