如何在Spark DataFrame中基于S3路径列生成JSON数据列（无需UDF）

完全可以实现，而且不需要使用UDF，下面给你两种实用方案：

方案一：批量读取JSON后关联原表 #

1. 先提取所有s3path列的路径，批量读取这些JSON文件：

// 提取所有S3路径到列表
val paths = df.select("s3path").as[String].collect()
// 批量读取JSON，同时用input_file_name()标记文件路径
val jsonDf = spark.read.json(paths)
  .withColumn("s3path", input_file_name())

2. 将原DataFrame和读取后的jsonDf按s3path列关联，就能得到每行对应JSON数据的结果：

val resultDf = df.join(jsonDf, Seq("s3path"), "left")

注意：如果各JSON文件结构不一致，需要提前开启Spark的schema合并配置：spark.sql.schemaMerge=true；如果文件数量极多，collect()可能触发内存溢出，建议分批次处理。

方案二：使用Spark 3.1+的read_file内置函数 #

Spark 3.1及以上版本提供了read_file函数，能直接在列表达式中读取指定路径的文件内容，搭配from_json就能解析成结构化数据：

import org.apache.spark.sql.functions.{read_file, from_json}
import org.apache.spark.sql.types._

// 预先定义JSON的schema（已知结构时推荐，性能更好）
val jsonSchema = StructType(Seq(
  StructField("field1", StringType),
  StructField("field2", IntegerType)
))

// 新增json_data列，存储解析后的JSON数据
val resultDf = df.withColumn(
  "json_data",
  from_json(read_file($"s3path"), jsonSchema)
)

如果JSON结构不确定，也可以省略schema参数让from_json自动推断，但性能会有所下降，且可能出现解析异常。

关键注意点 #

• 确保Spark集群拥有访问目标S3桶的权限（通过IAM角色或AWS凭证配置）
• 小文件数量较多时，方案二的性能更稳定，避免了批量读取的内存压力

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