Spark 和 Not Serialize DateTimeFormatter

serialization java scala apache-spark

2022-09-01 16:56:21

我正在尝试在Spark中使用java.time.format中的DateTimeFormatter，但它似乎不可序列化。这是相关的代码块：

val pattern = "<some pattern>".r
val dtFormatter = DateTimeFormatter.ofPattern("<some non-ISO pattern>")

val logs = sc.wholeTextFiles(path)

val entries = logs.flatMap(fileContent => {
    val file = fileContent._1
    val content = fileContent._2
    content.split("\\r?\\n").map(line => line match {
      case pattern(dt, ev, seq) => Some(LogEntry(LocalDateTime.parse(dt, dtFormatter), ev, seq.toInt))
      case _ => logger.error(s"Cannot parse $file: $line"); None
    })
  })

如何避免异常？有没有更好的库来解析时间戳？我读到Joda也是不可序列化的，并且已被合并到Java 8的时间库中。java.io.NotSerializableException: java.time.format.DateTimeFormatter

答案 1

您可以通过两种方式避免序列化：

假设其值可以是常量，请将格式化程序放在一个中（使其成为“静态”）。这意味着可以在每个工作线程中访问静态值，而不是驱动程序序列化它并发送到工作线程：object
```
object MyUtils {
  val dtFormatter = DateTimeFormatter.ofPattern("<some non-ISO pattern>")
}

import MyUtils._
logs.flatMap(fileContent => {
  // can safely use formatter here
})
```
在匿名函数中按记录实例化它。这会降低一些性能（因为实例化将按记录一遍又一遍地发生），因此仅当无法应用第一个选项时才使用此选项：
```
logs.flatMap(fileContent => {
  val dtFormatter = DateTimeFormatter.ofPattern("<some non-ISO pattern>")
  // use formatter here
})
```

答案 2

另一种方法是使DateTimeFormatter瞬态化。这告诉 JVM/Spark，该变量不是序列化的，而是从头开始构造的。对于每个执行器构建便宜的东西，比如DateTimeFormatter，这是一个好方法。

这是一篇更详细地描述这一点的文章。