Spark快速大数据分析（第2版）

122

｜

第

5

章

-- SQL查询

SELECT

id, collect_list(value + 1)

AS

values

FROM

(

SELECT

id, EXPLODE(values)

AS

value

FROM

table) x

GROUP BY

id

collect_list()

返回的是未去重的对象列表，由于

GROUP BY

语句会触发数据混洗操作，因

此重新组合的数组顺序和原数组不一定相同。因为

values

可能维度很高（是一个非常宽或

非常长的数组），而我们又在使用

GROUP BY

，所以这种方法可能开销巨大。

5.4.2

　方式

2

：

用户自定义函数

要想执行等价的任务（为

values

中的每个元素加

1

），也可以创建

UDF

，用

map()

迭代各

元素（

value

）并执行额外的操作。

val

plusOneInt = (values: Array[Int]) => {

values.map(value => value + 1)

}

// 注册UDF

spark.udf.register("plusOneInt", plusOneInt)

然后可以用如下方式在

Spark SQL

中使用这个

UDF

。

spark.sql("SELECT id, plusOneInt(values) AS values FROM table").show()

由于没有顺序问题，这种方法比使用

explode()

和

collect_list()

好一些，但序列化和反

序列化过程本身开销较大。同样值得注意的是，

collect_list() ...