如何在PySpark中读取JSON字符串格式文件并生成目标DataFrame

实现思路 #

因为文件每行的JSON字符串被额外的单引号包裹，无法直接使用PySpark内置的JSON读取接口，需要先做字符串预处理，再解析JSON结构：

1. 先以纯文本格式读取文件，每行对应DataFrame的一个字符串列
2. 去除每行首尾的单引号，得到标准JSON格式字符串
3. 定义JSON结构的Schema，解析处理后的字符串得到结构化数据
4. 展开解析后的结构体字段，得到最终的目标DataFrame

完整代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import regexp_extract, from_json
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 初始化SparkSession
spark = SparkSession.builder.appName("read_quoted_json").getOrCreate()

# 1. 定义JSON对应的Schema（显式定义比自动推断性能、稳定性更好）
json_schema = StructType([
    StructField("Name", StringType(), nullable=True),
    StructField("Age", IntegerType(), nullable=True)
])

# 2. 读取文本文件，处理每行的单引号
raw_df = spark.read.text("你的文件路径.txt")

# 3. 提取单引号中间的JSON内容，解析成结构化数据
processed_df = raw_df \
    .select(regexp_extract("value", r"'(.*)'", 1).alias("json_str")) \
    .select(from_json("json_str", json_schema).alias("data")) \
    .select("data.*")

# 查看结果
processed_df.show()

代码说明 #

• 提取单引号内容也可以用trim(col("value"), "'")替换regexp_extract，两种写法效果一致。
• 如果JSON结构复杂或者不确定，也可以让Spark自动推断Schema，不过生产环境更推荐显式定义Schema避免类型错误。

可选简化写法（基于RDD预处理） #

如果更习惯RDD操作，可以用更短的代码实现：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("read_quoted_json").getOrCreate()
# 读取文件后直接去除每行首尾单引号，再交给JSON接口解析
processed_df = spark.read.json(
    spark.sparkContext.textFile("你的文件路径.txt").map(lambda s: s.strip("'"))
)
processed_df.show()

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