sql - 在Spark-SQL中创建用户定义的函数

Question

我是Spark和Spark sql的新手，我试图使用Spark SQL查询一些数据。

我需要从以字符串形式给出的日期中获取月份。

我认为不可能直接从sparkqsl查询月份，所以我正在考虑在scala中编写用户定义的函数。

是否可以在sparkSQL中编写udf，并且如果可能的话，有人可以建议编写udf的最佳方法。

请帮忙

Answer 1

如果您愿意使用语言集成的查询，则至少可以进行过滤。

对于包含以下内容的数据文件date.txt：

one,2014-06-01
two,2014-07-01
three,2014-08-01
four,2014-08-15
five,2014-09-15

您可以根据需要在UDF中打包尽可能多的Scala日期魔术，但我将使其保持简单：

def myDateFilter(date: String) = date contains "-08-"

按以下步骤进行全部设置-其中很多来自 Programming guide。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import sqlContext._

// case class for your records
case class Entry(name: String, when: String)

// read and parse the data
val entries = sc.textFile("dates.txt").map(_.split(",")).map(e => Entry(e(0),e(1)))

您可以将UDF用作WHERE子句的一部分：

val augustEntries = entries.where('when)(myDateFilter).select('name, 'when)

并查看结果：

augustEntries.map(r => r(0)).collect().foreach(println)

请注意我使用的 where方法的版本，在文档中声明如下：

def where[T1](arg1: Symbol)(udf: (T1) ⇒ Boolean): SchemaRDD

因此，UDF只能接受一个参数，但是您可以编写多个 .where()调用以对多列进行过滤。

编辑Spark 1.2.0（实际上也是1.1.0）

虽然尚未真正记录，但Spark现在支持注册UDF，以便可以从SQL查询它。

可以使用以下命令注册以上UDF：

sqlContext.registerFunction("myDateFilter", myDateFilter)

如果表已注册

sqlContext.registerRDDAsTable(entries, "entries")

可以使用查询

sqlContext.sql("SELECT * FROM entries WHERE myDateFilter(when)")

有关更多详细信息，请参见 this example。