udf in Spark SQL DSL

Question

Я пытаюсь использовать DSL поверх чистого SQL в Spark sql-заданиях, но я не могу получить работу UDF.

sqlContext.udf.register("subdate",(dateTime: Long)=>dateTime.toString.dropRight(6)) 

Это не работает

rdd1.toDF.join(rdd2.toDF).where("subdate(rdd1(date_time)) === subdate(rdd2(dateTime))") 

Я также хотел бы добавить еще условие соединения, как в данном рабочем чистом SQL

val results=sqlContext.sql("select * from rdd1 join rdd2 on rdd1.id=rdd2.idand subdate(rdd1.date_time)=subdate(rdd2.dateTime)") 

Спасибо за вашу помощь

Answer 1

SQL Вы передаете метод where, неверно хотя бы по нескольким причинам:

• === - метод Column не является действительным равенством SQL. Вы должны использовать знак единственного равенства. =
• Обозначение скобок (table(column)) не является допустимым способом ссылки на столбцы в SQL. В этом контексте он будет распознан как вызов функции. SQL использует точечную нотацию (table.column)
• даже если она не была ни rdd1, ни rdd2 являются действительными псевдонимами таблицы

Поскольку это выглядит как имена столбцов являются однозначными вы могли бы просто использовать следующий код:

df1.join(df2).where("subdate(date_time) = subdate(dateTime)") 

Если бы это было не так, то использование синтаксиса точек не работало бы без предоставления псевдонимов в первую очередь. См., Например, Usage of spark DataFrame "as" method

Кроме того, регистрация UDF имеет смысл в основном при использовании необработанного SQL. Если вы хотите использовать DataFrame API, лучше использовать UDF напрямую:

import org.apache.spark.sql.functions.udf 

val subdate = udf((dateTime: Long) => dateTime.toString.dropRight(6)) 

val df1 = rdd1.toDF 
val df2 = rdd2.toDF 

df1.join(df2, subdate($"date_time") === subdate($"dateTime")) 

или если имена столбцов были неоднозначны:

df1.join(df2, subdate(df1("date_time")) === subdate(df2("date_time"))) 

Наконец, для простых функций, как это лучше создавать встроенные чем создавать UDF.