Apache Spark 如何将Map[String,String]转换为Map[String,Struct]类型

核心逻辑 #

你需要将col2（Map[String,String]类型）中每个值对应的JSON字符串解析为指定结构体，可直接通过Spark内置函数实现，无需自定义UDF，性能更高。

前置定义 #

首先声明内层结构体的Schema：

// Scala 定义内层Schema
import org.apache.spark.sql.types._
val innerSchema = StructType(Seq(
  StructField("SubCol1", StringType, nullable = true),
  StructField("SubCol2", StringType, nullable = true)
))

# PySpark 定义内层Schema
from pyspark.sql.types import StructType, StructField, StringType
inner_schema = StructType([
    StructField("SubCol1", StringType(), nullable=True),
    StructField("SubCol2", StringType(), nullable=True)
])

转换代码 #

Spark 3.1及以上版本（推荐） #

使用transform_values函数直接遍历Map的所有值做解析：

// Scala 实现
import org.apache.spark.sql.functions.{transform_values, from_json, col}

val resultDF = originalDF.withColumn("col2", 
  transform_values(col("col2"), (_, value) => from_json(value, innerSchema))
)

# PySpark 实现
from pyspark.sql.functions import transform_values, from_json, col

result_df = original_df.withColumn("col2",
  transform_values(col("col2"), lambda _, value: from_json(value, inner_schema))
)

Spark 3.1以下兼容版本 #

如果使用的Spark版本低于3.1，没有transform_values函数，可以通过拆分Map的键、值数组处理后再重组：

// Scala 兼容实现
import org.apache.spark.sql.functions.{map_from_arrays, map_keys, map_values, expr, col}

val resultDF = originalDF.withColumn("col2",
  map_from_arrays(
    map_keys(col("col2")),
    expr("transform(map_values(col2), x -> from_json(x, 'struct<SubCol1:string, SubCol2:string>'))")
  )
)

# PySpark 兼容实现
from pyspark.sql.functions import map_from_arrays, map_keys, map_values, col, expr

result_df = original_df.withColumn("col2",
  map_from_arrays(
    map_keys(col("col2")),
    expr("transform(map_values(col2), x -> from_json(x, 'struct<SubCol1:string, SubCol2:string>'))")
  )
)

结果验证 #

转换后的resultDF的Schema与你给出的目标Schema完全匹配，输出的JSON结构也符合预期要求。

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