如何用Pyspark实现混合普通列与JSON列的扁平文件数据扁平化处理

核心逻辑为三层处理：先拆分原始行得到独立的普通时间戳列，再通过内置JSON解析函数处理内嵌的JSON串为结构化列，最后提取目标字段合并为扁平化表，全程在DataFrame层完成，无需切换RDD写Map逻辑。

操作步骤 #

1. 按纯文本读取原始扁平文件，每行作为单独字符串存储

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import split, col, from_json
   from pyspark.sql.types import StructType, StructField, StringType
   
   spark = SparkSession.builder.appName("flatten_demo").getOrCreate()
   # 读取原始文件，每行内容存入默认的value列
   raw_df = spark.read.text("你的源文件存储路径")
   

2. 拆分每行的竖线分隔符，把普通时间戳和后续JSON字符串拆成两个独立列

   注意：竖线是正则特殊字符，拆分时需要转义，否则会出现拆分逻辑异常

   split_df = raw_df.select(
       split(col("value"), "\\|").getItem(0).alias("Timestamp"),
       split(col("value"), "\\|").getItem(1).alias("raw_json")
   )
   

3. 定义需要提取的JSON字段结构，把raw_json字符串解析为结构化列

   # 仅定义需要保留的JSON字段，无需纳入不需要的number字段
   json_schema = StructType([
       StructField("Colour", StringType()),
       StructField("Reason", StringType())
   ])
   parsed_df = split_df.withColumn("json_data", from_json(col("raw_json"), json_schema))
   

4. 从结构化JSON列中取出目标字段，与Timestamp列组合为最终扁平化结果

   final_df = parsed_df.select(
       col("Timestamp"),
       col("json_data.Colour"),
       col("json_data.Reason")
   )
   # 可直接展示查看结果
   final_df.show()
   

补充说明 #

• 原始文件行并非纯JSON格式，前置的时间戳片段会导致直接用spark.read.json解析失败，必须先拆分出独立的JSON片段再做结构化处理，这是拆分步骤存在的核心原因
• 这种实现方式相比手写RDD Map解析JSON的方法，底层由PySpark内置优化器执行，运行效率更高
• 若需要将Timestamp列转为标准时间类型，可直接在select阶段复用cast操作，和普通列处理逻辑完全一致，不存在合并障碍
• 若后续JSON字段有增减，只需要修改定义的json_schema结构即可，无需调整整体处理流程

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