Spark中移除StructType的Null/0字段以适配DynamoDB记录大小限制

核心思路 #

要在保留Map结构（而非JSON字符串）的同时移除Null字段，可通过将Struct转为Map并过滤Null值实现，既适配DynamoDB的类型要求，又能压缩数据大小。以下是两种可行方案：

方案1：自定义UDF过滤Null字段（兼容Spark 2.x及以上） #

如果使用较低版本的Spark，可自定义UDF将Struct类型转换为过滤掉Null值的Map：

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

// 定义UDF：将Struct转为过滤Null后的Map
val removeNullsFromStruct = udf((structRow: Row) => {
  structRow.schema.fields
    .zip(structRow.toSeq)
    .filter { case (_, value) => value != null }
    .map { case (field, value) => field.name -> value }
    .toMap
})

// 修正原代码语法错误，并应用UDF过滤Null
val resultDF = df
  // 先将0值转为Null（按需选择，若不需要保留0则启用）
  .withColumn("home_number", when(col("home_number") === 0, lit(null)).otherwise(col("home_number")))
  .withColumn("cell_number", when(col("cell_number") === 0, lit(null)).otherwise(col("cell_number")))
  .withColumn("work_number", when(col("work_number") === 0, lit(null)).otherwise(col("work_number")))
  // 构建Struct后转为过滤Null的Map
  .withColumn("phone_nos", struct(
    col("address"),
    col("home_number"),
    col("cell_number"),
    col("work_number")
  ))
  .withColumn("phone_nos", removeNullsFromStruct(col("phone_nos")))
  // 聚合操作
  .groupBy("name")
  .agg(collect_set(col("phone_nos")).alias("phone_nos_by_address"))

方案2：使用Spark内置函数（Spark 3.0+推荐） #

Spark 3.0+提供map_filter函数，Spark 3.3+提供to_map函数，无需UDF即可高效实现：

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

val resultDF = df
  // 按需将0值转为Null
  .withColumn("home_number", when(col("home_number") === 0, lit(null)).otherwise(col("home_number")))
  .withColumn("cell_number", when(col("cell_number") === 0, lit(null)).otherwise(col("cell_number")))
  .withColumn("work_number", when(col("work_number") === 0, lit(null)).otherwise(col("work_number")))
  // 将字段直接转为Map并过滤Null值
  .withColumn("phone_nos", map_filter(
    to_map(
      lit("address"), col("address"),
      lit("home_number"), col("home_number"),
      lit("cell_number"), col("cell_number"),
      lit("work_number"), col("work_number")
    ),
    (k, v) => v.isNotNull
  ))
  // 聚合操作
  .groupBy("name")
  .agg(collect_set(col("phone_nos")).alias("phone_nos_by_address"))

关键说明 #

• 若不需要将0转为Null，可删除when转换逻辑，直接保留0值；
• 处理后的phone_nos是Spark的Map[String, Any]类型，可直接映射到DynamoDB的Map类型，消费者无需解析JSON；
• 过滤Null字段后，每条记录的大小会显著降低，满足DynamoDB的记录大小限制。

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