Spark：UDF中的访问行答案

【问题标题】：Spark : Access Row inside an UDFSpark：UDF中的访问行
【发布时间】：2020-06-13 13:51:26
【问题描述】：

我有以下 UDF 用于将存储为字符串的时间转换为时间戳。

  val hmsToTimeStampUdf = udf((dt: String) => {
    if (dt == null) null else {
      val formatter = DateTimeFormat.forPattern("HH:mm:ss")
      try {
        new Timestamp(formatter.parseDateTime(dt).getMillis)
      } catch {
        case t: Throwable => throw  new RuntimeException("hmsToTimeStampUdf,dt="+dt, t)
      }
    }
  })

这个UDF用于将String值转换成Timestamp：

outputDf.withColumn(schemaColumn.name, ymdToTimeStampUdf(col(schemaColumn.name))

但某些 CSV 文件的此列的值无效，导致 RuntimeException。我想找出哪些行有这些损坏的记录。是否可以访问 UDF 中的行信息？

【问题讨论】：

嗯，不，但你可以有2个参数的UDF，时间列第二是ID列，然后打印它或其他东西。这个策略对你有用吗？

标签： apache-spark user-defined-functions

【解决方案1】：

与其抛出一个 RuntimeException 来杀死你的 .csv 解析，也许更好的方法是让 UDF 返回一个元组（格式正确，损坏的）值。然后，您可以通过选择 is null/is not null 子集轻松地分离好/坏行。

def safeConvert(dt: String) : (Timestamp,String) = {
  if (dt == null) 
    (null,null) 
  else {
    val formatter = DateTimeFormat.forPattern("HH:mm:ss")
    try {
      (new Timestamp(formatter.parseDateTime(dt).getMillis),null)
    } catch {
      case e:Exception => 
        (null,dt)
    }
  }
}
val safeConvertUDF = udf(safeConvert(_:String))

val df = Seq(("00:01:02"),("03:04:05"),("67:89:10")).toDF("dt")

df.withColumn("temp",safeConvertUDF($"dt"))
  .withColumn("goodData",$"temp".getItem("_1"))
  .withColumn("badData",$"temp".getItem("_2"))
  .drop($"temp").show(false)
+--------+-------------------+--------+
|dt      |goodData           |badData |
+--------+-------------------+--------+
|00:01:02|1970-01-01 00:01:02|null    |
|03:04:05|1970-01-01 03:04:05|null    |
|67:89:10|null               |67:89:10|
+--------+-------------------+--------+

【讨论】：

【解决方案2】：

您可以将该行作为第二个输入参数添加到 udf：

val hmsToTimeStampUdf = udf((dt: String, r: Row) => {
  if (dt == null) null else {
    val formatter = DateTimeFormat.forPattern("HH:mm:ss")
    try {
      new Timestamp(formatter.parseDateTime(dt).getMillis)
    } catch {
      case t: Throwable => {
        println(r) //do some error handling
        null
      }
    }
  }
})

调用 udf 时，使用带有数据帧所有列的结构作为第二个参数（感谢this answer）：

df.withColumn("dt", hmsToTimeStampUdf(col("dt"), struct(df.columns.map(df(_)) : _*)))

【讨论】：