Spark 2.0数据集：指定列转JSON后写入PostgreSQL方案问询

针对你的需求，我推荐两种实现方式——优先用Spark内置函数（简单高效），如果需要自定义逻辑再用UDF，下面详细说明：

方法一：使用Spark内置的struct + to_json函数（推荐） #

Spark已经内置了将结构体转为JSON字符串的函数，完全满足你"以原列名为键、列值为值"的需求，不需要额外引入Jackson或Lift库，代码简洁且性能更优。

示例代码（Scala） #

import org.apache.spark.sql.functions.{struct, to_json}

// 从原DataFrame中选择6列，将剩余4列打包为JSON字符串
val ds_transformed = ds_base
  .select(
    $"bill_id",
    $"trans_id",
    $"billing_id",
    $"asset_id",
    $"row_id",
    $"created",
    // 将end_dt、start_dt、status_cd、update_start_dt转为JSON，别名为additional_data
    to_json(struct($"end_dt", $"start_dt", $"status_cd", $"update_start_dt")).alias("additional_data")
  )

原理说明 #

• struct(...)：把指定的4列打包成一个Spark结构体对象，保留原列名作为结构体的字段名。
• to_json(...)：自动将结构体序列化为JSON字符串，最终生成的JSON格式就是{"end_dt":"xxx", "start_dt":"xxx", ...}，完全符合你的要求。

方法二：自定义UDF（适合需要特殊JSON处理的场景） #

如果你因为某些原因必须用自定义UDF（比如修改键名、特殊空值处理），可以通过接收多参数或Row对象来实现——你之前失败大概率是因为Jackson没有正确配置Scala模块，下面是修正后的代码：

示例代码（Scala + Jackson） #

import org.apache.spark.sql.functions.udf
import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import scala.collection.mutable.Map

// 初始化Jackson ObjectMapper，必须注册Scala模块才能正确序列化Scala集合
val mapper = new ObjectMapper()
mapper.registerModule(DefaultScalaModule)

// 定义UDF，接收4个列值参数，返回JSON字符串
val colsToJson = udf((endDt: String, startDt: String, statusCd: String, updateStartDt: String) => {
  // 构建键值对映射，键用原列名
  val dataMap = Map(
    "end_dt" -> endDt,
    "start_dt" -> startDt,
    "status_cd" -> statusCd,
    "update_start_dt" -> updateStartDt
  )
  // 序列化为JSON字符串
  mapper.writeValueAsString(dataMap)
})

// 使用UDF转换数据
val ds_transformed = ds_base
  .select(
    $"bill_id",
    $"trans_id",
    $"billing_id",
    $"asset_id",
    $"row_id",
    $"created",
    colsToJson($"end_dt", $"start_dt", $"status_cd", $"update_start_dt").alias("additional_data")
  )

写入PostgreSQL的后续步骤 #

转换完成后，就可以用JDBC将ds_transformed写入目标表了，注意目标表的第7列（比如additional_data）建议定义为JSONB类型（PostgreSQL对JSONB支持更好，可建索引），或者TEXT类型：

ds_transformed.write
  .format("jdbc")
  .option("url", "jdbc:postgresql://your-db-host:5432/your-db-name")
  .option("dbtable", "your-target-table-name")
  .option("user", "your-username")
  .option("password", "your-password")
  .mode("append") // 根据需求选择模式：append/overwrite/ignore等
  .save()

额外提示 #

• 空值处理：to_json会自动将空列值序列化为null，如果需要忽略空值，可以在struct中用when函数过滤，比如struct(when($"end_dt".isNotNull, $"end_dt") as "end_dt", ...)。
• 性能对比：内置to_json函数是Spark优化过的，比自定义UDF性能高很多，所以优先使用内置函数。

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