在Spark SQL中的一个查询中使用多个collect_list

Question

我有以下数据帧data:

root
 |-- userId: string 
 |-- product: string 
 |-- rating: double

以及以下查询:

val result = sqlContext.sql("select userId, collect_list(product), collect_list(rating) from data group by userId")

我的问题是,不product和rating在聚集阵列相互匹配？也就是说,同一行product和rating来自同一行的聚合数组中是否具有相同的索引.

更新:从Spark 2.0.0开始,可以collect_list对struct类型执行,因此我们可以collect_list在组合列上执行一个.但是对于2.0.0之前的版本,人们只能使用collect_list原始类型.

Answer 1

我相信没有明确保证所有阵列都具有相同的顺序.Spark SQL使用多个优化,并且在某些条件下不能保证所有聚合同时被调度(一个示例是聚合DISTINCT).由于交换(混洗)导致不确定的顺序,理论上订单可能会有所不同.

因此,尽管它应该在实践中起作用,但它可能存在风险并且会引入一些难以发现的错误.

如果您使用Spark 2.0.0或更高版本,则可以使用以下方法聚合非原子列collect_list:

SELECT userId, collect_list(struct(product, rating)) FROM data GROUP BY userId

如果您使用的是早期版本,则可以尝试使用显式分区并订购:

WITH tmp AS (
  SELECT * FROM data DISTRIBUTE BY userId SORT BY userId, product, rating
)
SELECT userId, collect_list(product), collect_list(rating)
FROM tmp
GROUP BY userId