Spark如何将DataFrame所有列合并为单列JSON且保留类型不转义引号

问题根因 #

转义问题的核心是：存储JSON内容的列当前是String类型，Spark执行JSON序列化时会将其识别为普通字符串，自动转义内部的双引号。要解决这个问题，只需提前将所有存储JSON字符串的列解析为Spark结构化类型，再整体序列化即可。

通用解决方案（无需手动定义Schema，支持多列批量处理） #

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

// 批量处理所有字符串列，自动解析JSON内容为结构化类型
val parsedDF = someDF.columns.foldLeft(someDF) { (tempDF, colName) =>
  // 仅处理字符串类型列，可自定义过滤规则指定需要解析的JSON列提升性能
  if (tempDF.schema(colName).dataType == StringType) {
    // 采样首行非空值自动推导JSON Schema，无需手动定义
    val sampleJson = tempDF.select(colName).na.drop.limit(1).as[String].headOption
    sampleJson match {
      // 仅处理合法的JSON对象/数组格式字符串
      case Some(jsonStr) if jsonStr.trim.startsWith("{") || jsonStr.trim.startsWith("[") =>
        val jsonSchema = schema_of_json(lit(jsonStr))
        tempDF.withColumn(colName, from_json(col(colName), jsonSchema))
      case _ => tempDF
    }
  } else {
    tempDF
  }
}

// 整体序列化得到无转义的JSON结果
val resultDF = parsedDF.select(to_json(struct(col("*"))).alias("value"))

// 输出验证
resultDF.show(false)

输出结果 #

+--------------------------------------------------------------------+
|value                                                               |
+--------------------------------------------------------------------+
|{"number":8,"word":{"details":{"decision":"ACCEPT","source":"Rules"}}}|
|{"number":64,"word":{"details":{"decision":"ACCEPT","source":"Rules"}}}|
+--------------------------------------------------------------------+

注意事项 #

• 如果已经明确知道哪些列是JSON列，可直接指定列名列表转换，不需要全列扫描，性能更高
• 如果JSON结构存在行级不一致的情况，可给from_json传入options(Map("mode" -> "PERMISSIVE"))参数，兼容格式异常的JSON数据
• 采样推导Schema时如果首行是空值会自动跳过该列，也可调整采样行数（比如取前100行）获取更准确的Schema

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