Spark2.3.2/Hive替换数组结构体Null值及写入分区表实现

一、Spark Scala 实现方案（适配Spark 2.3.2） #

由于Spark 2.3.2不支持transform函数，且表中列数超过64列，直接explode后分组会触发列数限制报错，因此我们直接针对Students结构体数组字段编写UDF处理Null值，无需改动其他80+列。

1. 定义结构体与UDF #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 请根据表中Students结构体的实际字段调整该case class
case class Student(id: Int, score: Int, name: String)

// 创建处理结构体数组的UDF：替换id、score的Null值为0
val fixStudentNulls = udf((students: Seq[Student]) => {
  students.map { s =>
    Student(
      id = if (s.id == null) 0 else s.id,
      score = if (s.score == null) 0 else s.score,
      name = s.name // 其他字段原样保留
    )
  }
})

2. 读取Hive表并处理数据 #

val spark = SparkSession.builder()
  .appName("FixStudentNulls")
  .enableHiveSupport()
  .getOrCreate()

// 读取分区表，保留所有原始列，仅修改Students字段
val df = spark.table("student_details")
  .withColumn("Students", fixStudentNulls(col("Students")))

// 写入分区Hive表（覆盖或追加模式根据需求选择）
df.write
  .mode("overwrite") // 或 "append"
  .partitionBy("Report_Date") // 按原分区字段分区
  .saveAsTable("student_details_fixed")

二、Hive SQL 实现方案 #

通过posexplode展开数组并保留元素索引，替换Null值后重新聚合为数组，避免因列数过多导致的分组报错。

执行SQL（需替换为实际列名） #

-- 插入到修复后的分区表
INSERT OVERWRITE TABLE student_details_fixed PARTITION(Report_Date)
SELECT 
  -- 替换为表中实际的80+列名，排除Students字段
  col1, col2, ..., col80,
  collect_list(named_struct(
    'id', coalesce(s.id, 0),
    'score', coalesce(s.score, 0),
    'name', s.name -- 其他结构体字段原样保留
  )) AS Students,
  Report_Date
FROM student_details
LATERAL VIEW posexplode(Students) exploded AS idx, s
GROUP BY 
  col1, col2, ..., col80, Report_Date, idx -- 通过idx保证数组元素顺序
ORDER BY idx;

注意：若无需严格保留数组元素顺序，可去掉idx相关的分组与排序，仅按原始列和Report_Date分组即可。

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