PySpark如何按event_name为单数据集不同JSON行应用对应Schema解析

你遇到的报错分两层：

• 表层是语法错误：PySpark的when函数多条件需要链式调用，你第二个when没有接在第一个when的返回结果后面，语法不合法。
• 深层是逻辑问题：即使修正语法，也会报错，因为不同event_name对应的JSON Schema不同，from_json返回的Struct结构不一致，Spark不允许同一个列存储不同结构的Struct类型数据，这才是核心问题。

下面给出3种常用的生产环境方案，可根据你的场景选择：

方案1：分事件类型解析后合并（最推荐，性能最优） #

适合事件类型较多、后续需要按事件类型独立分析聚合的场景：

1. 提前定义所有事件类型的Schema映射
2. 按事件类型拆分数据集，分别解析JSON、扁平化字段
3. 统一字段后合并为全量表
   代码示例：

from pyspark.sql.types import *
import pyspark.sql.functions as F

# 第一步：定义所有事件的Schema映射
event_schema_map = {
    "EventStart": StructType([
        StructField("Name", StringType()),
        StructField("Version", IntegerType()),
        StructField("Id", IntegerType())
    ]),
    "Action1": StructType([
        StructField("Name", StringType()),
        StructField("Version", IntegerType()),
        StructField("UserName", StringType()),
        StructField("PosX", IntegerType()),
        StructField("PosY", IntegerType())
    ])
    # 其他事件Schema继续追加
}

# 第二步：获取所有公共字段（除了json_string之外的常规字段）
common_cols = [c for c in df.columns if c != "json_string"]
all_dfs = []

# 第三步：逐个处理每种事件
for event_name, schema in event_schema_map.items():
    # 过滤对应事件的子数据集
    event_df = df.filter(F.col("event_name") == event_name)
    # 解析JSON并扁平化所有字段
    event_df = event_df.withColumn("json_parsed", F.from_json("json_string", schema)) \
                       .select(*common_cols, "json_parsed.*")
    all_dfs.append(event_df)

# 第四步：合并所有事件的数据集，缺失字段自动填充null
final_df = all_dfs[0]
for next_df in all_dfs[1:]:
    final_df = final_df.unionByName(next_df, allowMissingColumns=True)

方案2：生成事件专属字段（适合事件类型少的场景） #

不需要合并数据集，直接将不同事件的JSON字段解析为带事件名前缀的独立列，生成宽表：

df = df \
  # 解析EventStart字段
  .withColumn("EventStart_json", F.when(F.col("event_name") == "EventStart", F.from_json("json_string", "Name String, Version Int, Id Int"))) \
  .select("*", "EventStart_json.*") \
  .drop("EventStart_json") \
  # 解析Action1字段
  .withColumn("Action1_json", F.when(F.col("event_name") == "Action1", F.from_json("json_string", "Name String, Version Int, UserName String, PosX Int, PosY Int"))) \
  .select("*", "Action1_json.*") \
  .drop("Action1_json")

非对应事件的专属字段会自动返回null，可直接用于后续聚合。

方案3：按需提取JSON字段（适合临时分析，无需预定义Schema） #

如果你只需要用到JSON中的部分字段，不需要全量解析，可以直接用get_json_object提取，不需要关心事件Schema：

df = df \
  .withColumn("Name", F.get_json_object("json_string", "$.Name")) \
  .withColumn("Version", F.get_json_object("json_string", "$.Version").cast(IntegerType())) \
  .withColumn("UserName", F.get_json_object("json_string", "$.UserName")) \
  .withColumn("PosX", F.get_json_object("json_string", "$.PosX").cast(IntegerType()))
  # 其他需要的字段继续追加即可

字段不存在的行自动返回null，无需处理事件匹配逻辑。

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