Azure Databricks作业日志非标准JSON格式优雅提取问询

从Azure Databricks生成的作业日志JSON中提取数据时，遇到部分字段为带转义引号的类字典格式字符串，示例如下：

"response": "{\"statusCode\":200}"

这类字符串因首尾双引号和内部转义符的存在，无法直接解析为标准JSON。目前使用Scala构建日志的层级映射结构，可通过点符号正常访问properties.sourceIPAddress这类字段，但正则预处理未成功，希望找到无需逐行循环的优雅内联提取方式。

以下是几种基于Spark DataFrame的Scala内联处理方案，无需逐行遍历：

方案一：字符串清洗 + from_json解析结构体 #

先通过字符串函数清理格式问题，再用from_json将字符串解析为结构化数据：

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

// 定义目标字段的Schema
val responseSchema = StructType(Seq(
  StructField("statusCode", IntegerType, nullable = true)
))

val processedDF = originalDF
  // 去掉首尾双引号，再移除内部转义反斜杠
  .withColumn("cleaned_response", 
    regexp_replace(regexp_replace(col("response"), "^\"|\"$", ""), "\\\\", "")
  )
  // 将清洗后的字符串解析为结构体
  .withColumn("response_struct", from_json(col("cleaned_response"), responseSchema))

// 直接通过点符号访问解析后的字段
processedDF.select("response_struct.statusCode").show()

方案二：get_json_object直接提取单字段 #

若仅需提取字符串中的特定字段，无需完整解析结构体，可直接使用get_json_object：

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

val processedDF = originalDF
  .withColumn("response_status", 
    get_json_object(
      // 先去掉首尾双引号
      regexp_replace(col("response"), "^\"|\"$", ""), 
      // 指定JSON路径提取字段
      "$.statusCode"
    ).cast(IntegerType)
  )

processedDF.select("response_status").show()

方案三：自定义UDF处理复杂格式 #

针对更复杂的非标准格式，可编写自定义UDF实现灵活解析：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import scala.util.Try
import org.json4s._
import org.json4s.jackson.JsonMethods._

// 自定义UDF：清洗字符串并解析为Map
val parseResponseUdf = udf((responseStr: String) => {
  Try {
    val cleanedStr = responseStr.stripPrefix("\"").stripSuffix("\"").replace("\\", "")
    parse(cleanedStr).extract[Map[String, Any]]
  }.toOption
})

val processedDF = originalDF
  .withColumn("response_map", parseResponseUdf(col("response")))

// 提取目标字段
processedDF.select(col("response_map.statusCode").cast(IntegerType)).show()

注：Databricks环境默认包含json4s依赖，无需额外引入。

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