Spark中如何将嵌套结构化字符串转换为Struct类型

问题背景 #

有一个包含多条JSON记录的文件，每条记录包含名为Properties的Struct，其中的meterDetails字段是如下格式的非标准字符串：

"@{meterName=Read Operations; meterCategory=Storage; meterSubCategory=General Block Blob; unitOfMeasure=100000000}"

注意：字符串中的键值未用双引号包裹，也不符合JSON格式规范。尝试通过regexp_replace移除@后用from_json转换、或通过split转数组再转JSON，结果新列全为NULL，疑问是否需要将该字符串转为Spark可原生生成Struct的对象。

核心原因 #

你之前的方法失效，是因为原字符串不是标准JSON格式：

• JSON要求键和值必须用双引号包裹（字符串类型）
• 键值对之间需用逗号分隔，而非分号
• 仅移除@后得到的{meterName=...}仍不符合JSON语法，from_json无法解析

解决步骤 #

必须先将非标准字符串转换为标准JSON格式，再用from_json解析为Struct。以下是具体实现（以Scala为例）：

1. 定义目标Struct的Schema #

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

val meterDetailsSchema = StructType(Seq(
  StructField("meterName", StringType),
  StructField("meterCategory", StringType),
  StructField("meterSubCategory", StringType),
  StructField("unitOfMeasure", StringType)
))

2. 转换字符串为标准JSON #

通过多层regexp_replace完成格式修正：

val processedDf = df
  // 提取Properties中的meterDetails字段，去掉开头的@{和结尾的}
  .withColumn("cleaned_str", regexp_replace(regexp_replace(col("Properties.meterDetails"), "^@\\{", ""), "\\}$", ""))
  // 将key=value替换为"key":"value"，给键和值加上双引号
  .withColumn("json_str", regexp_replace(col("cleaned_str"), "([^=;]+)=([^;]+)", "\"$1\":\"$2\""))
  // 将分号分隔替换为逗号分隔
  .withColumn("standard_json", regexp_replace(col("json_str"), "; ", ", "))

3. 解析JSON为Struct并展开 #

// 解析标准JSON为Struct
val withStructDf = processedDf.withColumn("meterDetails_struct", from_json(col("standard_json"), meterDetailsSchema))

// 展开Struct的所有字段到DataFrame中
val finalDf = withStructDf
  .select("*", "meterDetails_struct.*")
  .drop("cleaned_str", "json_str", "standard_json", "meterDetails_struct")

替代方案（手动构建Struct） #

如果正则表达式处理复杂，也可以通过拆分字符串手动构建Struct：

val finalDf = df
  .withColumn("meter_kv", split(regexp_replace(col("Properties.meterDetails"), "^@\\{|\\}$", ""), "; "))
  .withColumn("meter_map", map_from_entries(
    transform(col("meter_kv"), kv => split(kv, "="))
  ))
  .select(
    "*",
    col("meter_map.meterName").alias("meterName"),
    col("meter_map.meterCategory").alias("meterCategory"),
    col("meter_map.meterSubCategory").alias("meterSubCategory"),
    col("meter_map.unitOfMeasure").alias("unitOfMeasure")
  )
  .drop("meter_kv", "meter_map")

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