如何在PySpark中删除列中的空列表并对全表列执行explode操作

核心思路 #

你要的效果本质是对每列单独执行过滤空列表、展开操作，补全其他列为空值后按原列顺序合并所有结果。

实现代码 #

from pyspark.sql import functions as F

# 第一步：过滤所有列均为空列表的无效行
# 生成过滤条件：只要任意一列非空就保留该行
filter_condition = F.greatest(*[F.size(col_name) > 0 for col_name in pivotTest.columns])
filtered_df = pivotTest.filter(filter_condition)

# 第二步：批量处理每一列
processed_dfs = []
for current_col in filtered_df.columns:
    # 单独处理当前列：过滤当前列非空的行、执行explode展开
    single_col_result = filtered_df.select(current_col) \
        .filter(F.size(current_col) > 0) \
        .select(F.explode(current_col).alias(current_col))
    # 补全其余所有列为空值，保证和原表列顺序、列数一致
    for other_col in filtered_df.columns:
        if other_col != current_col:
            single_col_result = single_col_result.withColumn(other_col, F.lit(None))
    # 对齐列顺序
    single_col_result = single_col_result.select(filtered_df.columns)
    processed_dfs.append(single_col_result)

# 第三步：合并所有列的处理结果
final_df = processed_dfs[0]
for df in processed_dfs[1:]:
    final_df = final_df.unionByName(df)

# 查看结果
final_df.show()

代码说明 #

• 过滤全空行的逻辑适配任意列数的DataFrame，不需要手动指定列名
• 批量处理逻辑自动遍历所有列，新增列时不需要修改代码
• 全程自动对齐原表列顺序，不会出现列错位问题
• 完全兼容Spark 3.1.2 + Python 3 运行环境

内容的提问来源于stack exchange，提问作者Jesús M. Carrillo M.