scala - Spark UDF 线程安全

Question

我正在使用 Spark 获取包含一列日期的数据框，并创建 3 个新列，其中包含该列中的日期与今天之间的天数、周数和月数。

我担心的是 SimpleDateFormat 的使用，它不是线程安全的。通常没有 Spark 这会很好，因为它是一个局部变量，但是使用 Spark 的懒惰评估，在多个 UDF 上共享单个 SimpleDateFormat 实例可能会导致问题吗？

def calcTimeDifference(...){
    val sdf = new SimpleDateFormat(dateFormat)

    val dayDifference = udf{(x: String) => math.abs(Days.daysBetween(new DateTime(sdf.parse(x)), presentDate).getDays)}
    output = output.withColumn("days", dayDifference(myCol))

    val weekDifference = udf{(x: String) => math.abs(Weeks.weeksBetween(new DateTime(sdf.parse(x)), presentDate).getWeeks)}
    output = output.withColumn("weeks", weekDifference(myCol))

    val monthDifference = udf{(x: String) => math.abs(Months.monthsBetween(new DateTime(sdf.parse(x)), presentDate).getMonths)}
    output = output.withColumn("months", monthDifference(myCol))
}

Answer 1

我认为它不安全，正如我们所知，简单日期格式 不是线程安全的。

所以我更喜欢这种方法使用 简单日期格式 如果需要，在 Spark 中:

import java.text.SimpleDateFormat
import java.util.SimpleTimeZone

/**
  * Thread Safe SimpleDateFormat for Spark.
  */
object ThreadSafeFormat extends ThreadLocal[SimpleDateFormat] {

  override def initialValue(): SimpleDateFormat = {
    val dateFormat = new SimpleDateFormat("yyyy-MM-dd:H")
    // if you need get UTC time, you can set UTC timezone
    val utcTimeZone = new SimpleTimeZone(SimpleTimeZone.UTC_TIME, "UTC")
    dateFormat.setTimeZone(utcTimeZone)
    dateFormat
  }

}

然后使用 ThreadSafeFormat.get()让线程安全的 SimpleDateFormat 做任何事情。