我是Spark和Spark sql的新手,我试图使用Spark SQL查询一些数据。
我需要从以字符串形式给出的日期中获取月份。
我认为不可能直接从sparkqsl查询月份,所以我正在考虑在scala中编写用户定义的函数。
是否可以在sparkSQL中编写udf,并且如果可能的话,有人可以建议编写udf的最佳方法。
请帮忙
最佳答案
如果您愿意使用语言集成的查询,则至少可以进行过滤。
对于包含以下内容的数据文件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。
关于sql - 在Spark-SQL中创建用户定义的函数,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/25031129/