Scala Spark udf java.lang.UnsupportedOperationException

Question

Я создал эту функцию каррирования для проверки на наличие нулевых значений для endDateStr внутри udf, код выглядит следующим образом: (Тип столбца x - ArrayType [TimestampType]):

    def _getCountAll(dates: Seq[Timestamp]) = Option(dates).map(_.length)
    def _getCountFiltered(endDate: Timestamp)(dates: Seq[Timestamp]) = Option(dates).map(_.count(!_.after(endDate)))

    val getCountUDF = udf((endDateStr: Option[String]) => {
      endDateStr match {
        case None => _getCountAll _
        case Some(value) => _getCountFiltered(Timestamp.valueOf(value + " 23:59:59")) _
      }
    })
    df.withColumn("distinct_dx_count", getCountUDF(lit("2009-09-10"))(col("x")))

НоЯ получаю это исключение при выполнении:

java.lang.UnsupportedOperationException: схема для типа Seq [java.sql.Timestamp] => Параметр [Int] не поддерживается

Может кто-нибудь помочь мне разобраться в моей ошибке?

Answer 1

Вы не можете карри udf, как это.Если вы хотите поведение, похожее на карри, вы должны вернуть udf из внешней функции:

def getCountUDF(endDateStr: Option[String]) = udf {
  endDateStr match {
    case None => _getCountAll _
    case Some(value) => 
      _getCountFiltered(Timestamp.valueOf(value + " 23:59:59")) _
  }
}

df.withColumn("distinct_dx_count", getCountUDF(Some("2009-09-10"))(col("x")))

, в противном случае просто отбросьте карри и предоставьте оба аргумента одновременно:

val getCountUDF = udf((endDateStr: String, dates: Seq[Timestamp]) => 
  endDateStr match {
    case null => _getCountAll(dates)
    case _ => 
      _getCountFiltered(Timestamp.valueOf(endDateStr + " 23:59:59"))(dates)
  }
)

df.withColumn("distinct_dx_count", getCountUDF(lit("2009-09-10"), col("x")))