文档章节

Spark UDF变长参数的二三事儿

问津已非少年
 问津已非少年
发布于 2017/07/04 19:43
字数 1351
阅读 781
收藏 0

引子

变长参数对于我们来说并不陌生,在Java里我们这么写

public void varArgs(String... args)

在Scala里我们这么写

def varArgs(cols: String*): String

而在Spark里,很多时候我们有自己的业务逻辑,现成的functions满足不了我们的需求,而当我们需要处理同一行的多个列,将其经过我们自己的逻辑合并为一个列时,变长参数及其变种实现可以给我们提供帮助。

但是在Spark UDF里我们是 无法使用变长参数传值 的,但之所以本文以变长参数开头,是因为需求起于它,而通过对它进行变换,我们可以使用变长参数或Seq类型来接收参数。

下面通过Spark-Shell来做演示,以下三种方法都可以做到多列传参,分别是

变长参数类型的UDF

定义UDF方法

def myConcatVarargs(sep: String, cols: String*): String = cols.filter(_ != null).mkString(sep)

注册UDF函数

由于变长参数只能通过方法定义,所以这里使用部分应用函数来转换

val myConcatVarargsUDF = udf(myConcatVarargs _)

可以看到该UDF的定义如下

UserDefinedFunction(<function2>,StringType,List(StringType, ArrayType(StringType,true)))

也即变长参数转换为了ArrayType,而且函数是只包括两个参数,所以变长参数列表由此也可看出无法使用的。

变长参数列表传值

我们构造一个DataFrame如下

val df = sc.parallelize(Array(("aa", "bb", "cc"),("dd","ee","ff"))).toDF("A", "B", "C")

然后直接传入多个String类型的列到myConcatVarargsUDF

df.select(myConcatVarargsUDF(lit("-"), col("A"), col("B"), col("C"))).show

结果出现如下报错

java.lang.ClassCastException: anonfun$1 cannot be cast to scala.Function4

由此可以看出,使用变长参数列表的方式Spark是不支持的,它会被识别为四个参数的函数,而UDF确是被定义为两个参数而不是四个参数的函数!

变换:使用array()转换做第二个参数

我们使用Spark提供的array() function来转换参数为Array类型

df.select(myConcatVarargsUDF(lit("-"), array(col("A"), col("B"), col("C")))).show

结果如下

+-------------------+
|UDF(-,array(A,B,C))|
+-------------------+
|           aa-bb-cc|
|           dd-ee-ff|
+-------------------+

由此可以看出,使用变长参数构造的UDF方法,可以通过构造Array的方式传参,来达到多列合并的目的。

使用Seq类型参数的UDF

上面提到,变长参数最后被转为ArrayType,那不禁要想我们为嘛不使用Array或List类型呢?

实际上在UDF里,类型并不是我们可以随意定义的,比如使用List和Array就是不行的,我们自己定义的类型也是不行的,因为这涉及到数据的序列化和反序列化。

以Array/List为示例的错误

下面以Array类型为示例

定义函数

val myConcatArray = (cols: Array[String], sep: String) => cols.filter(_ != null).mkString(sep)

注册UDF

val myConcatArrayUDF = udf(myConcatArray)

可以看到给出的UDF签名是

UserDefinedFunction(<function2>,StringType,List())

应用UDF

df.select(myConcatArrayUDF(array(col("A"), col("B"), col("C")), lit("-"))).show

会发现报错

scala.collection.mutable.WrappedArray$ofRef cannot be cast to [Ljava.lang.String

同样List作为参数类型也会报错,因为反序列化的时候无法构建对象,所以List和Array是无法直接作为UDF的参数类型的

以Seq做参数类型

定义调用如下

val myConcatSeq = (cols: Seq[Any], sep: String) => cols.filter(_ != null).mkString(sep)

val myConcatSeqUDF = udf(myConcatSeq)

df.select(myConcatSeqUDF(array(col("A"), col("B"), col("C")), lit("-"))).show

结果如下

+-------------------+
|UDF(array(A,B,C),-)|
+-------------------+
|           aa-bb-cc|
|           dd-ee-ff|
+-------------------+

使用Row类型参数的UDF

我们可以使用Spark functions里struct方法构造结构体类型传参,然后用Row类型接UDF的参数,以达到多列传值的目的。

def myConcatRow: ((Row, String) => String) = (row, sep) => row.toSeq.filter(_ != null).mkString(sep)

val myConcatRowUDF = udf(myConcatRow)

df.select(myConcatRowUDF(struct(col("A"), col("B"), col("C")), lit("-"))).show

可以看到UDF的签名如下

UserDefinedFunction(<function2>,StringType,List())

结果如下

+--------------------+
|UDF(struct(A,B,C),-)|
+--------------------+
|            aa-bb-cc|
|            dd-ee-ff|
+--------------------+

使用Row类型还可以使用模式提取,用起来会更方便

row match {
  case Row(aa:String, bb:Int) =>
}

最后

对于上面三种方法,变长参数和Seq类型参数都需要array的函数包装为ArrayType,而使用Row类型的话,则需要struct函数构建结构体类型,其实都是为了数据的序列化和反序列化。三种方法中,Row的方式更灵活可靠,而且支持不同类型并且可以明确使用模式提取,用起来相当方便。

而由此我们也可以看出,UDF不支持List和Array类型的参数,同时 自定义参数类型 如果没有混合Spark的特质实现序列化和反序列化,那么在UDF里也是 无法用作参数类型 的。当然,Seq类型是可以 的,可以接多列的数组传值。

此外,我们也可以使用柯里化来达到多列传参的目的,只是不同参数个数需要定义不同的UDF了。

欢迎阅读转载,转载请注明出处:https://my.oschina.net/u/2539801/blog/1154536

© 著作权归作者所有

问津已非少年
粉丝 18
博文 21
码字总数 33944
作品 0
海淀
程序员
私信 提问
Apache Spark 2.4 正式发布,重要功能详细介绍

美国时间 2018年11月08日 正式发布了。一如既往,为了继续实现 Spark 更快,更轻松,更智能的目标,Spark 2.4 带来了许多新功能,如下: 添加一种支持屏障模式(barrier mode)的调度器,以便...

Spark
2018/11/10
0
0
Spark - 动态注册UDF

昨天有位大哥问小弟一个Spark问题,他们想在不停Spark程序的情况下动态更新UDF的逻辑,他一问我这个问题的时候,本猪心里一惊,Spark**还能这么玩?我出于程序员的本能回复他肯定不行,但今天...

大猪大猪
05/24
0
0
Apache Spark 2.4.0 正式发布

Apache Spark 2.4 与昨天正式发布,Apache Spark 2.4 版本是 2.x 系列的第五个版本。 如果想及时了解 Spark、Hadoop或者Hbase相关的文章,欢迎关注微信公共帐号: itebloghadoop Apache Spa...

Spark
2018/11/09
0
0
Spark - udf动态代码注册

昨天有位大哥问小弟一个Spark问题,他们想在不停Spark程序的情况下动态更新UDF的逻辑,他一问我这个问题的时候,本猪心里一惊,Spark**还能这么玩?我出于程序员的本能回复他肯定不行,但今天...

大猪大猪
05/07
0
0
Spark2.x写入Elasticsearch的性能测试

一、Spark集成ElasticSearch的设计动机 ElasticSearch 毫秒级的查询响应时间还是很惊艳的。其优点有: 1. 优秀的全文检索能力 2. 高效的列式存储与查询能力 3. 数据分布式存储(Shard 分片) 相...

openfea
2017/10/27
336
0

没有更多内容

加载失败,请刷新页面

加载更多

rime设置为默认简体

转载 https://github.com/ModerRAS/ModerRAS.github.io/blob/master/_posts/2018-11-07-rime%E8%AE%BE%E7%BD%AE%E4%B8%BA%E9%BB%98%E8%AE%A4%E7%AE%80%E4%BD%93.md 写在开始 我的Arch Linux上......

zhenruyan
今天
5
0
简述TCP的流量控制与拥塞控制

1. TCP流量控制 流量控制就是让发送方的发送速率不要太快,要让接收方来的及接收。 原理是通过确认报文中窗口字段来控制发送方的发送速率,发送方的发送窗口大小不能超过接收方给出窗口大小。...

鏡花水月
今天
9
0
OSChina 周日乱弹 —— 别问,问就是没空

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @tom_tdhzz :#今日歌曲推荐# 分享容祖儿/彭羚的单曲《心淡》: 《心淡》- 容祖儿/彭羚 手机党少年们想听歌,请使劲儿戳(这里) @wqp0010 :周...

小小编辑
今天
942
11
golang微服务框架go-micro 入门笔记2.1 micro工具之micro api

micro api micro 功能非常强大,本文将详细阐述micro api 命令行的功能 重要的事情说3次 本文全部代码https://idea.techidea8.com/open/idea.shtml?id=6 本文全部代码https://idea.techidea8....

非正式解决方案
今天
5
0
Spring Context 你真的懂了吗

今天介绍一下大家常见的一个单词 context 应该怎么去理解,正确的理解它有助于我们学习 spring 以及计算机系统中的其他知识。 1. context 是什么 我们经常在编程中见到 context 这个单词,当...

Java知其所以然
昨天
7
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部