Scala中RDD日期验证、时间戳转换及无效行过滤方案

需求背景 #

现有RDD[(String, String)]类型数据，每个字符串元素为yyyy-MM-dd HH:mm:ss格式的日期时间戳。原通过SimpleDateFormat实现的epochTime函数仅做转换，未处理空值、空字符串或格式错误的情况，需要修改逻辑，过滤掉包含无效值的行，仅保留并转换有效行的时间戳。

原函数代码：

def epochTime (stringOfTime: String): Long = dateFormats.parse(stringOfTime).getTime

解决方案 #

1. 重构时间转换函数（支持异常与空值处理） #

将函数返回值改为Option[Long]，以此标记转换是否成功，同时处理空值、空串和格式错误：

import java.text.SimpleDateFormat
import java.util.Date

// SimpleDateFormat非线程安全，Spark分布式环境下用ThreadLocal隔离
private val dateFormat = new ThreadLocal[SimpleDateFormat] {
  override def initialValue(): SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
}

def epochTime(stringOfTime: String): Option[Long] = {
  Option(stringOfTime) // 处理null值
    .filter(_.trim.nonEmpty) // 过滤空串或全空格字符串
    .flatMap { timeStr =>
      try {
        // 解析成功则返回时间戳
        Some(dateFormat.get.parse(timeStr).getTime)
      } catch {
        // 捕获格式错误等异常，返回None标记无效
        case _: Exception => None
      }
    }
}

2. 处理RDD过滤有效行 #

通过flatMap遍历RDD，仅保留两个字段都能成功转换的行：

val originalRDD: RDD[(String, String)] = ... // 你的原始RDD实例

val validEpochRDD: RDD[(Long, Long)] = originalRDD
  .flatMap { case (time1, time2) =>
    // 仅当两个时间都转换成功时，返回元组
    for {
      epoch1 <- epochTime(time1)
      epoch2 <- epochTime(time2)
    } yield (epoch1, epoch2)
  }

3. 推荐：使用Java 8+时间API替代SimpleDateFormat #

SimpleDateFormat已过时且线程不安全，Spark 2.0+环境下推荐用java.time包的API，无需ThreadLocal，更可靠：

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.format.DateTimeParseException

// DateTimeFormatter是线程安全的，可全局复用
private val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")

def epochTime(stringOfTime: String): Option[Long] = {
  Option(stringOfTime)
    .filter(_.trim.nonEmpty)
    .flatMap { timeStr =>
      try {
        // 解析为LocalDateTime，转成UTC毫秒级时间戳（与原getTime结果一致）
        Some(LocalDateTime.parse(timeStr, formatter).toEpochSecond(java.time.ZoneOffset.UTC) * 1000)
      } catch {
        case _: DateTimeParseException => None
      }
    }
}

输入输出示例 #

输入 #

(2020-10-10 05:17:12,2015-04-10 09:18:20)
(2020-10-12 06:15:58,2015-04-10 09:17:42)
(2020-10-11 07:16:40,2015-04-10 09:17:49)

输出 #

(1602303432,1428653900)
(1602479758,1428653862)
(1602397000,1428653869)

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