pyspark - PySpark Dataframe 转置为列表
问题描述
我正在使用 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
将group_id
andpyspark.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)
struct 方法的有用之处在于您可以使用点访问器按名称访问嵌套数组的元素:
df2.select("group_id", "members.member_id").show()
#+--------+----------+
#|group_id| member_id|
#+--------+----------+
#| 65| [345]|
#| 55|[123, 234]|
#+--------+----------+
推荐阅读
- c++ - 如何在 C++ 代码块中从 oracle 数据库中读取表?我
- opencv - 如何配置 while 循环以使用 picamera?
- angular - Angular 单元测试为 Http Post 返回“无法读取未定义的属性”
- mysql - 如何在 django2 中编写 raw like 查询?
- javascript - Javascript变量传递带有撇号的文本字符串,被切断了吗?
- c++ - C++ 隐式类到 char* 的转换
- python - 组合不同的列
- java - 使用 Kotlin DSL 配置 Java 规范的 Gradle 编译错误
- c# - C# 编写目录
- java - 有什么简单的解释说明修改 Java catch 子句中的异常参数 ex 的充分理由吗?