Scala UDF завершается неудачно при вызове из оператора SELECT в DataBricks / Spark

У меня есть следующая функция Scala, которую я зарегистрировал как UDF в Azure DataBricks (Spark 2.3.1 и Scala 2.11):

import org.joda.time.DateTime

val slot = (dt : DateTime) => { 
  var _s : Int = (dt.dayOfYear().get() - 1) * 24 + dt.hourOfDay().get()
  _s  
}

val compute_slot = (t: String, offset: Int) => {
  var dt:DateTime = DateTime.parse(t) 
  ((Set(slot(dt)) ++ Set(slot(dt.minusSeconds(offset)))) ++ Set(slot    (dt.plusSeconds(offset)))).toArray  
}

    spark.udf.register("get_slot", compute_slot)

Функция работает хорошо, если вызывается напрямую...

compute_slot("2018-11-01T05:04:33.827+0000", 1800)

... или как UDF в SQL:

%sql
SELECT explode(get_slot(cast("2018-11-01T05:04:33.827+0000" as string), 1800))

При использовании UDF в SQL-запросе SparkSql…

%sql
SELECT * FROM tab LATERAL VIEW explode(get_slot(cast(timestamp as string), 1800)) my_view

… или же

val ColUDF = udf(compute_slot)
df.withColumn("arr", ColUDF($"timestamp", lit(1800))).show()

это терпит неудачу со следующим сообщением об ошибке. В нем упоминается неправильный формат даты, который смущает меня, так как функция хорошо работает вне оператора SELECT над моей таблицей.

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 177.0 failed 4 times, most recent failure: Lost task 0.3 in stage 177.0 (TID 33992, 10.139.64.4, executor 4): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$2: (string, int) => array<int>)

at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:620)
    at org.apache.spark.sql.execution.collect.UnsafeRowBatchUtils$.encodeUnsafeRows(UnsafeRowBatchUtils.scala:49)
    at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:126)
    at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:125)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:112)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:384)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Invalid format: "2018-11-01 05:04:33.827" is malformed at " 05:04:33.827"
    at org.joda.time.format.DateTimeFormatter.parseDateTime(DateTimeFormatter.java:945)
    at org.joda.time.DateTime.parse(DateTime.java:160)
    at org.joda.time.DateTime.parse(DateTime.java:149)
    at line7c3ca3974ac14b88a9a351882d40a949342.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$2.apply(command-1611582591913421:11)
    at line7c3ca3974ac14b88a9a351882d40a949342.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$2.apply(command-1611582591913421:10)
    ... 12 more

Любая помощь, чтобы сосредоточиться на том, что я делаю неправильно, будет высоко ценится!

Большое спасибо!

Paolo

Источник

2 ответа

Решение

По умолчанию parse использования ISODateTimeFormat с шаблоном, определенным как:

datetime          = time | date-opt-time
time              = 'T' time-element [offset]
date-opt-time     = date-element ['T' [time-element] [offset]]
date-element      = std-date-element | ord-date-element | week-date-element
std-date-element  = yyyy ['-' MM ['-' dd]]
ord-date-element  = yyyy ['-' DDD]
week-date-element = xxxx '-W' ww ['-' e]
time-element      = HH [minute-element] | [fraction]
minute-element    = ':' mm [second-element] | [fraction]
second-element    = ':' ss [fraction]
fraction          = ('.' | ',') digit+
offset            = 'Z' | (('+' | '-') HH [':' mm [':' ss [('.' | ',') SSS]]])

пока ваша дата не содержит offset составная часть. Чтобы соответствовать 2018-11-01 05:04:33.827 Вы должны предоставить формат:

scala> import org.joda.time.format.DateTimeFormat
import org.joda.time.format.DateTimeFormat
scala> DateTime.parse("2018-11-01 05:04:33.827", DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS"));
res3: org.joda.time.DateTime = 2018-11-01T05:04:33.827+01:00
Источник

Колонка timestamp неправильно отформатирован. Это говорит об ошибке:

Причина: java.lang.IllegalArgumentException: неверный формат: "2018-11-01 05:04:33.827" искажен в " 05:04:33.827"

Кроме того, это не удалось:

get_slot(cast(timestamp as string), 1800)

Но это удалось:

get_slot(cast("2018-11-01T05:04:33.827+0000" as string), 1800)

Так что ваши timestamp столбец просто не отформатирован должным образом. Просто добавьте T часть, и ты будешь в порядке.

Источник
Другие вопросы по тегам