Spark UDTF 与 输出多列的UDF小笔记

原创
2017/07/21 16:56
阅读数 3.5K

前言

对于UDF和UDAF相比大家并不陌生,用起来也都很顺手,在此就不多做介绍了~

通常我们的UDF都是一个或多个输入,然后一个输出。如果想要使用多个输出,比如像1.6版本里的json_tuple这样 :

df.select(json_tuple(col("jsonField"), "Field1", "Field2").as(Seq("Field1", "Field2")))

这种输出方式可以直接输出多列,同时 explode 也是类似的,这一类Spark内置函数称为 UDTF 也即 User Defined Table-generation Function,函数的输入是一个或多个列值,输出是一个table

不过Spark目前并未提供我们自己实现UDTF的方法,所以目前的使用还仅限于Spark内置的几个UDTF,想要达到多列输出,我们需要略作变通

使用Seq做返回类型,输出的多列是一个数组

def test: (String => Seq[String]) = aa => Seq("aa", aa, "cc")

这时候 udf(test) 的返回类型就是 ArrayType ,通过Column的getItem方法,我们可以取数得到多列

val testUDF = udf(test)
df.select(testUDF(col("testCol")).as("testUDFCol"))
  .select(col("testUDFCol").getItem(0).as("col1"),
    col("testUDFCol").getItem(1).as("col2"))

使用元组或case class做返回类型

case class Test(id: String, name: String)
def test: (String => Test) = aa => Test("1", aa)

// 使用元组做返回类型
def test: (String => (String, String)) = aa => ("1", aa)

当使用元组或case class做函数返回类型时,UDF的返回类型就是StructType,通过Column的getField方法,我们可以取数得到多列

val testUDF = udf(test)
df.select(testUDF(col("testCol")).as("testUDFCol"))
  .select(col("testUDFCol").getField("id").as("col1"),
    col("testUDFCol").getField("name").as("col2"))

// 使用元组的时候取值
df.select(testUDF(col("testCol")).as("testUDFCol"))
  .select(col("testUDFCol").getField("_1").as("col1"),
    col("testUDFCol").getField("_2").as("col2"))

使用Map作为返回类型

def test: (String => Map[String, String]) = aa => Map("id" -> "1", "name" -> aa)

对于Map类型,UDF的返回类型是MapType,也是通过Column的getField方法,我们可以取数得到多列

val testUDF = udf(test)
df.select(testUDF(col("testCol")).as("testUDFCol"))
  .select(col("testUDFCol").getField("id").as("col1"),
    col("testUDFCol").getField("name").as("col2"))

最后

想说的是,这种变通方式好像跟输出多列的UDF这个标题扯得有点远了,其实并没有输出多列,只是换一个类型而已,但问题由此引起,也就以此为题啦~~

另外,显然使用case class这种方式更好,因为返回类型有多种,更适合实际项目需要~~

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部