PySpark如何创建字面量数组列并对DataFrame执行explode操作

解决步骤 #

你之前的写法未生效大概率是没有提前导入PySpark对应的内置函数，完整可运行的实现逻辑如下：

1. 导入依赖函数 #

首先从pyspark.sql.functions导入需要用到的array、lit、explode三个函数：

from pyspark.sql.functions import array, lit, explode

2. 生成带数组列的DataFrame #

你原来的数组列创建逻辑本身是正确的，补全导入后就能正常生成每一行值都为["source1", "source2", "source3"]的source列：

# 你的原始DataFrame
df = spark.createDataFrame([(100, 'AB', 304), (200, 'BC', 305), (300, 'CD', 306)], ['number', 'letter', 'id'])
# 新增数组类型的source列
df_arr = df.withColumn("source", array(lit("source1"), lit("source2"), lit("source3")))

此时查看df_arr的结构，source列已经是ArrayType(StringType)的标准数组类型。

3. 执行explode操作 #

直接调用explode函数对source列做展开即可，每一行的数组元素会拆分成独立行，其他列的值会自动复制：

df_exploded = df_arr.withColumn("source", explode("source"))

最终运行结果示例 #

执行df_exploded.show()的输出如下：

+------+------+---+-------+
|number|letter| id| source|
+------+------+---+-------+
|   100|    AB|304|source1|
|   100|    AB|304|source2|
|   100|    AB|304|source3|
|   200|    BC|305|source1|
|   200|    BC|305|source2|
|   200|    BC|305|source3|
|   300|    CD|306|source1|
|   300|    CD|306|source2|
|   300|    CD|306|source3|
+------+------+---+-------+

额外说明 #

你之前尝试用numpy数组失败是因为PySpark的内置函数不能直接识别numpy数组类型，所有列值的生成需要用PySpark原生的lit等内置函数包装，才能被Spark的序列化机制识别。

内容的提问来源于stack exchange，提问作者viji