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

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

0 голосов
/ 05 декабря 2018

У меня есть следующая функция 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

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

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

Паоло

Ответы [ 2 ]

0 голосов
/ 05 декабря 2018

Столбец 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 часть, и все будет хорошо.

0 голосов
/ 05 декабря 2018

По умолчанию 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
...