Spark/Scala DataFrame groupBy聚合时如何忽略null值合并数据

问题场景 #

使用Scala做Spark开发时，直接调用groupBy做DataFrame聚合，无法实现忽略字段null值、按顺序合并同组非空值的效果。

• 待处理测试DataFrame包含JoB、Code1、Code2、Code3、ID共5个字段
• 同一ID+JoB分组下，Code1/Code2/Code3的非空值分散在不同行，其余对应字段位置均为null
  原始数据如下：

+----------+--------------+-----------+-----------+--------+
|JoB       |         Code1|      Code2|      Code3|      ID|
+----------+--------------+-----------+-----------+--------+
|         0|           Haa|       null|       null|      85|
|         0|          null|         HN|       null|      85|
|         0|          null|       null|        191|      85|
|         0|          null|         CE|       null|      85|
|         0|           HRa|       null|       null|      85|
|         0|          null|       null|        348|      85|
|         0|          null|       null|        203|      85|
|         1|           UlS|       null|       null|      85|
|         0|          null|         HV|       null|      85|
|         1|          null|       null|        840|      85|
|         1|          null|         UA|       null|      85|
|         0|           ClZ|       null|       null|      85|
+----------+--------------+-----------+-----------+--------+

期望输出 #

同组下各字段非空值按出现顺序配对，合并为无null的多行数据，结果如下：

+----------+--------------+-----------+-----------+--------+
|JoB       |         Code1|      Code2|      Code3|      ID|
+----------+--------------+-----------+-----------+--------+
|         0|           Haa|         HN|        191|      85|
|         0|           HRa|         CE|        348|      85|
|         1|           UlS|         UA|        840|      85|
|         0|           ClZ|         HV|        203|      85|
+----------+--------------+-----------+-----------+--------+

实现方案 #

常规groupBy聚合会将同组数据合并为单行，无法实现多列非空值按顺序对齐的需求，需要先给各字段非空值按分组打顺序编号，再通过编号关联合并。

1. 导入依赖

   import org.apache.spark.sql.expressions.Window
   import org.apache.spark.sql.functions._
   

2. 分别提取三个Code字段的非空值，按JoB、ID分组，按原始读取顺序生成行号

   // 提取Code1非空值并生成组内序号
   val code1Df = df.filter(col("Code1").isNotNull)
     .withColumn("rn", row_number().over(Window.partitionBy("JoB", "ID").orderBy(lit(1))))
     .select("JoB", "ID", "Code1", "rn")
   
   // 提取Code2非空值并生成组内序号
   val code2Df = df.filter(col("Code2").isNotNull)
     .withColumn("rn", row_number().over(Window.partitionBy("JoB", "ID").orderBy(lit(1))))
     .select("JoB", "ID", "Code2", "rn")
   
   // 提取Code3非空值并生成组内序号
   val code3Df = df.filter(col("Code3").isNotNull)
     .withColumn("rn", row_number().over(Window.partitionBy("JoB", "ID").orderBy(lit(1))))
     .select("JoB", "ID", "Code3", "rn")
   

   代码中orderBy(lit(1))默认按Spark读取原始数据的顺序编号，如果业务要求按特定字段（如时间戳、自增ID）排序，替换lit(1)为对应字段即可。

3. 按分组字段+行号做全外连接，合并三列非空值，删除辅助序号列

   val result = code1Df.join(code2Df, Seq("JoB", "ID", "rn"), "outer")
     .join(code3Df, Seq("JoB", "ID", "rn"), "outer")
     .drop("rn")
     .orderBy("JoB", "ID")
   

注意事项 #

• 若不同Code字段同组下非空值数量不一致，外连接会保留所有行，缺失位置自动补null，可根据业务需求加过滤规则处理
• 执行result.show(false)即可查看和预期一致的输出结果

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