选项1:使用
选项2:使用
我正在使用PySpark SQL API,并尝试将重复值的行分组为其余内容列表。它类似于转置,但不是旋转所有值,而是将值放入数组中。
当前输出:
group_id | member_id | name
55 | 123 | jake
55 | 234 | tim
65 | 345 | chris
所需的输出:
group_id | members
55 | [[123, 'jake'], [234, 'tim']]
65 | [345, 'chris']
您需要groupby
CC_1并使用pyspark.sql.functions.collect_list()
作为聚合功能。
对于组合member_id
和name
列,您有两个选择:
选项1:使用pyspark.sql.functions.array
:
from pyspark.sql.functions import array, collect_list
df1 = df.groupBy("group_id")
.agg(collect_list(array("member_id", "name")).alias("members"))
df1.show(truncate=False)
#+--------+-------------------------------------------------+
#|group_id|members |
#+--------+-------------------------------------------------+
#|55 |[WrappedArray(123, jake), WrappedArray(234, tim)]|
#|65 |[WrappedArray(345, chris)] |
#+--------+-------------------------------------------------+
这返回了字符串阵列的WrappedArray
。整数被转换为字符串,因为您不能混合类型数组。
df1.printSchema()
#root
# |-- group_id: integer (nullable = true)
# |-- members: array (nullable = true)
# | |-- element: array (containsNull = true)
# | | |-- element: string (containsNull = true)
选项2:使用pyspark.sql.functions.struct
from pyspark.sql.functions import collect_list, struct
df2 = df.groupBy("group_id")
.agg(collect_list(struct("member_id", "name")).alias("members"))
df2.show(truncate=False)
#+--------+-----------------------+
#|group_id|members |
#+--------+-----------------------+
#|65 |[[345,chris]] |
#|55 |[[123,jake], [234,tim]]|
#+--------+-----------------------+
这将返回一系列结构,并带有member_id
和name
df2.printSchema()
#root
# |-- group_id: integer (nullable = true)
# |-- members: array (nullable = true)
# | |-- element: struct (containsNull = true)
# | | |-- member_id: integer (nullable = true)
# | | |-- name: string (nullable = true)
对结构方法有用的是,您可以使用点访问器访问嵌套数组的元素:
df2.select("group_id", "members.member_id").show()
#+--------+----------+
#|group_id| member_id|
#+--------+----------+
#| 65| [345]|
#| 55|[123, 234]|
#+--------+----------+