PySpark：如何处理Delta表中JSON字段的类型不一致问题

针对Delta表中JSON列嵌套字段value同时存在字符串、数字、null类型，导致Structured Streaming解析时丢失数据的问题，以下是无需修改原始payload的高效解决方案：

方案1：使用parse_json统一转换为字符串（Spark 3.x+ 推荐） #

Spark 3.x及以上版本的parse_json支持动态解析JSON，无需提前指定Schema，提取value后直接转为字符串类型即可保留所有数据：

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

val streamDF = spark.readStream
  .format("delta")
  .load("/path/to/your/delta-table")
  // 解析JSON列并提取value字段
  .withColumn("parsed_value", parse_json($"json_col").getField("value"))
  // 统一转换为字符串类型，数字、字符串、null均保留
  .withColumn("unified_value", $"parsed_value".cast(StringType))

处理后unified_value列会将数字456转为"456"，字符串"123"保留为"123"，null仍为null，无数据丢失。

方案2：get_json_object结合条件处理（兼容低版本Spark） #

如果使用Spark 3.x以下版本，可通过get_json_object提取value的原始JSON表述，再去除字符串的引号：

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

val streamDF = spark.readStream
  .format("delta")
  .load("/path/to/your/delta-table")
  // 提取value的原始JSON片段
  .withColumn("raw_value", get_json_object($"json_col", "$.value"))
  // 去除字符串的引号，数字和null直接保留
  .withColumn("unified_value", 
    when($"raw_value".startsWith("\""), regexp_replace($"raw_value", "^\"|\"$", ""))
    .otherwise($"raw_value")
  )

方案3：自定义轻量UDF（兼容所有Spark版本） #

若上述方法不适用，可基于Jackson库编写UDF，高效解析并统一返回字符串：

import org.apache.spark.sql.functions._
import com.fasterxml.jackson.databind.ObjectMapper

// 定义UDF：解析JSON并提取value转为字符串
val parseValueUdf = udf((jsonStr: String) => {
  val mapper = new ObjectMapper()
  val jsonNode = mapper.readTree(jsonStr)
  val valueNode = jsonNode.get("value")
  if (valueNode == null || valueNode.isNull) null else valueNode.asText()
})

val streamDF = spark.readStream
  .format("delta")
  .load("/path/to/your/delta-table")
  .withColumn("unified_value", parseValueUdf($"json_col"))

注意事项 #

• 流处理中使用parse_json动态解析时，若Delta表存在Schema变更风险，需开启mergeSchema选项：.option("mergeSchema", "true")
• 所有方案均仅在读取阶段处理数据，无需修改原始Delta表的payload，完全符合性能要求

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