Spark集群环境下JSON列值分词代码适配问题求助

原代码在集群环境失败的核心原因：

• UDF运行在Executor节点，但代码中直接调用Driver端的spark.sql()并执行collect()，这会导致：
  1. SparkSession对象无法序列化到Executor，抛出序列化异常；
  2. 每个JSON字段值都触发一次独立的SQL查询，性能极差且严重违反Spark分布式计算模型；
  3. 字符串拼接SQL存在注入风险，特殊字符会导致语法错误。

1. 注册Hive UDF为Spark可调用的UDF #

先将Hive的mmtok函数注册到Spark，避免在Executor中触发独立SQL查询：

// 方式1：直接通过类路径注册临时函数（替换为你的UDF实际包路径）
spark.sql("CREATE TEMPORARY FUNCTION mmtok AS 'com.your.package.MmtokUDF'")

// 方式2：通过代码实例化UDF并注册（适用于自定义UDF）
spark.udf.register("mmtok", (value: String, strategy: String) => {
  val udfInstance = new com.your.package.MmtokUDF()
  udfInstance.evaluate(value, strategy).toString
})

2. 重写JSON分词逻辑，规避Driver端对象依赖 #

使用Play JSON处理嵌套JSON，直接调用注册好的mmtok函数，全程在Executor本地执行：

import play.api.libs.json._

// 获取注册好的mmtok函数本地引用
val mmtokFunc = spark.udf.lookup("mmtok").get.asInstanceOf[(String, String) => String]

// 递归遍历JSON，对所有字符串值分词
def tokenizeJson(jsonValue: JValue): JValue = jsonValue match {
  case JString(value) if value.trim.nonEmpty => JString(mmtokFunc(value, "myfunc"))
  case JString(value) => JString("") // 空字符串保持原样
  case JObject(fields) => JObject(fields.map { case (key, value) =>
    key -> tokenizeJson(value)
  })
  case other => other // 非字符串/对象类型（如JNull、JNumber）保持原样
}

// 定义处理JSON字符串的UDF
val tokenizeJsonUDF = udf((jsonStr: String) => {
  try {
    val parsedJson = Json.parse(jsonStr)
    val tokenizedJson = tokenizeJson(parsedJson)
    Json.stringify(tokenizedJson)
  } catch {
    case e: Exception =>
      // 异常时返回原JSON，避免数据丢失
      println(s"JSON处理错误: ${e.getMessage}")
      jsonStr
  }
})

3. 处理DataFrame生成结果 #

// 示例数据
val myStr = """{
  "name": "John",
  "address": {
    "street": "5th Avenue",
    "city": {
      "name": "New York",
      "details": {
        "population": "8 million",
        "area": "783.8 km²"
      }
    }
  }
}"""

val data = Seq(("John", "Business", myStr, "US", "Travel"))
val myDF = data.toDF("name", "profession", "personData", "country", "hobby")

// 应用UDF处理personData列
val finalDF = myDF.withColumn("personData", tokenizeJsonUDF(col("personData")))

finalDF.show(false)

额外优化：固定JSON结构时的高效处理 #

如果JSON结构固定，建议解析为StructType后逐个字段处理，性能比递归处理任意JSON更高：

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

// 定义JSON对应的Schema
val personSchema = StructType(Seq(
  StructField("name", StringType),
  StructField("address", StructType(Seq(
    StructField("street", StringType),
    StructField("city", StructType(Seq(
      StructField("name", StringType),
      StructField("details", StructType(Seq(
        StructField("population", StringType),
        StructField("area", StringType)
      )))
    )))
  )))
)

// 解析JSON为Struct，逐个字段分词后重新拼接为JSON
val tokenizedDF = myDF
  .withColumn("parsedData", from_json(col("personData"), personSchema))
  .withColumn("tokenized_name", callUDF("mmtok", col("parsedData.name"), lit("myfunc")))
  .withColumn("tokenized_street", callUDF("mmtok", col("parsedData.address.street"), lit("myfunc")))
  .withColumn("tokenized_city_name", callUDF("mmtok", col("parsedData.address.city.name"), lit("myfunc")))
  .withColumn("tokenized_population", callUDF("mmtok", col("parsedData.address.city.details.population"), lit("myfunc")))
  .withColumn("tokenized_area", callUDF("mmtok", col("parsedData.address.city.details.area"), lit("myfunc")))
  .withColumn("personData", to_json(struct(
    col("tokenized_name").alias("name"),
    struct(
      col("tokenized_street").alias("street"),
      struct(
        col("tokenized_city_name").alias("name"),
        struct(
          col("tokenized_population").alias("population"),
          col("tokenized_area").alias("area")
        ).alias("details")
      ).alias("city")
    ).alias("address")
  )))
  .drop("parsedData", "tokenized_name", "tokenized_street", "tokenized_city_name", "tokenized_population", "tokenized_area")

tokenizedDF.show(false)

• 确保Hive UDF的JAR包已通过--jars参数或集群配置提交到所有节点；
• 绝对禁止在Executor中访问Driver端对象（如SparkSession、SparkContext）；
• 递归处理任意JSON时需注意性能，大数量级数据优先使用固定Schema的处理方式；
• 异常处理避免返回空结构，建议返回原JSON以保证数据完整性。

内容的提问来源于stack exchange，提问作者Manoj Kumar G