PySpark中将包含空值的列值拆分展开为多行的问题求助

问题根因 #

出现该问题的核心原因有两个常见可能：

1. 拆分后的数组存在null值：如果原列值为null，split函数返回结果为null，arrays_zip传入null参数时返回的整体数组也为null，inline会直接过滤掉对应行，inline_outer则会返回全空的字段值。
2. 三个拆分后数组长度不一致：arrays_zip默认按最短数组的长度对齐，若某行存在长度为0的空数组，也会导致最终展开结果为空。

解决方案 #

优先处理null值并对齐数组长度后再执行展开操作，参考代码如下：

import pyspark.sql.functions as F

df = (df
  # 拆分列同时处理null值，空值转换为空数组
  .withColumn("col1_arr", F.coalesce(F.split("col1", ","), F.array()))
  .withColumn("col2_arr", F.coalesce(F.split("col2", ","), F.array()))
  .withColumn("col3_arr", F.coalesce(F.split("col3", ","), F.array()))
  # 计算当前行三个数组的最大长度
  .withColumn("max_len", F.greatest(F.size("col1_arr"), F.size("col2_arr"), F.size("col3_arr")))
  # 所有数组补全到最大长度，缺位补null
  .withColumn("col1_arr", F.expr("""
    if(size(col1_arr) < max_len, 
      concat(col1_arr, array_repeat(null, max_len - size(col1_arr))), 
      col1_arr
    )
  """))
  .withColumn("col2_arr", F.expr("""
    if(size(col2_arr) < max_len, 
      concat(col2_arr, array_repeat(null, max_len - size(col2_arr))), 
      col2_arr
    )
  """))
  .withColumn("col3_arr", F.expr("""
    if(size(col3_arr) < max_len, 
      concat(col3_arr, array_repeat(null, max_len - size(col3_arr))), 
      col3_arr
    )
  """))
  # 数组压缩后展开
  .withColumn("arr", F.arrays_zip("col1_arr", "col2_arr", "col3_arr"))
  .selectExpr("id", "inline_outer(arr)")
  # 重命名列到预期名称
  .withColumnRenamed("col1_arr", "col1")
  .withColumnRenamed("col2_arr", "col2")
  .withColumnRenamed("col3_arr", "col3")
)
df.show(truncate=False)

可选优化 #

如果原始数据的逗号分隔符前后存在空格，拆分后可对每个元素做去空格处理，示例如下：

transform(split(col1, ','), x -> trim(x)) as col1_arr

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