如何处理嵌套MapType对象：字符串转数组并清理指定键？

需求说明 #

• 处理嵌套MapType结构，若内层Map存在reports键：
  1. 将其对应的JSON字符串解析为数组
  2. 移除数组内每个对象的name键后，重新转为JSON字符串
• 保留其他所有键值对不变

列Schema #

|-- col_name: map (nullable = true)
 |    |-- key: string
 |    |-- value: map (valueContainsNull = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)

输入输出示例 #

输入 #

{
  'h#9l00' : {
     'reports': '[{"name":"abc1", "id": "emp1"}, {"name":"abc2", "id": "emp2"}]',
     'salary': 100,
     'city': 'xyz'
  },
  'o*5ftr': {
     'city': 'pqr',
     'salary': 1000
  }
} 

输出 #

{
  'h#9l00' : {
     'reports': '[{"id": "emp1"}, {"id": "emp2"}]',
     'salary': 100,
     'city': 'xyz'
  },
  'o*5ftr': {
     'city': 'pqr',
     'salary': 1000
  }
} 

实现方案 #

以Spark为例，通过自定义UDF处理嵌套Map：

Python 版本 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.types import MapType, StringType
import json

# 定义处理内层Map的函数
def process_inner_map(inner_map):
    if not inner_map or 'reports' not in inner_map:
        return inner_map
    
    try:
        # 解析JSON字符串为列表
        reports_list = json.loads(inner_map['reports'])
        # 移除每个元素的name键
        cleaned_reports = [{k: v for k, v in item.items() if k != 'name'} for item in reports_list]
        # 转回JSON字符串并替换原字段
        inner_map['reports'] = json.dumps(cleaned_reports)
    except Exception as e:
        # 处理JSON解析异常，保留原数据
        pass
    return inner_map

# 注册UDF
process_inner_map_udf = udf(process_inner_map, MapType(StringType(), StringType()))

# 应用到目标列
df = df.withColumn('col_name', process_inner_map_udf(col('col_name')))

Scala 版本 #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import scala.util.parsing.json.JSON
import scala.util.Try

// 定义处理内层Map的函数
def processInnerMap(innerMap: Map[String, String]): Map[String, String] = {
    innerMap.get("reports") match {
        case Some(reportsStr) =>
            Try {
                val reportsList = JSON.parseFull(reportsStr).get.asInstanceOf[List[Map[String, String]]]
                val cleanedReports = reportsList.map(_ - "name")
                innerMap + ("reports" -> JSON.compact(cleanedReports))
            }.getOrElse(innerMap)
        case None => innerMap
    }
}

// 注册UDF
val processInnerMapUdf = udf(processInnerMap, MapType(StringType, StringType))

// 应用到目标列
val df = df.withColumn("col_name", processInnerMapUdf(col("col_name")))

注意事项 #

• 加入了异常处理，避免因reports格式不合法导致任务失败
• 若reports字段可能为null，函数已做空值判断，确保逻辑健壮性

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