PySpark嵌套JSON扁平化：如何避免值重复实现目标输出

问题描述 #

输入JSON文件内容：

{"events": [{"event_name": "start","event_properties": ["property1", "property2", "property3"],"entities": ["entityI", "entityII", "entityIII"],"event_timestamp": "2022-05-01 00:00:00"},{"event_name": "stop","event_properties": ["propertyA", "propertyB", "propertyC", "propertyD"],"entities": ["entityW", "entityX", "entityY", "entityZ"],"event_timestamp": "2022-05-01 01:00:00"}]}

期望输出格式：

用户尝试的代码：

# Importing package
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StructField, StringType, IntegerType,BooleanType,DoubleType

# Implementing JSON File in PySpark

spark = SparkSession.builder \
    .master("local[1]") \
    .appName("PySpark Read JSON") \
    .getOrCreate()
    
df = spark.read.option("multiline","true").json(r"C:\Users\Lajo\Downloads\spark_ex1_input.json")

from pyspark.sql.types import *
from pyspark.sql.functions import explode_outer,col


def flatten(df):
   # compute Complex Fields (Lists and Structs) in Schema   
   complex_fields = dict([(field.name, field.dataType)
                             for field in df.schema.fields
                             if type(field.dataType) == ArrayType or  type(field.dataType) == StructType])
   while len(complex_fields)!=0:
      col_name=list(complex_fields.keys())[0]
      print ("Processing :"+col_name+" Type : "+str(type(complex_fields[col_name])))
    
      # if StructType then convert all sub element to columns.
      # i.e. flatten structs
      if (type(complex_fields[col_name]) == StructType):
         expanded = [col(col_name+'.'+k).alias(col_name+'_'+k) for k in [ n.name for n in  complex_fields[col_name]]]
         df=df.select("*", *expanded).drop(col_name)
    
      # if ArrayType then add the Array Elements as Rows using the explode function
      # i.e. explode Arrays
      elif (type(complex_fields[col_name]) == ArrayType):    
         df=df.withColumn(col_name,explode_outer(col_name))
    
      # recompute remaining Complex Fields in Schema       
      complex_fields = dict([(field.name, field.dataType)
                             for field in df.schema.fields
                             if type(field.dataType) == ArrayType or  type(field.dataType) == StructType])
   return df
   
   
 df_flatten = flatten(df)
 
 df_flatten.show()

问题分析 #

原代码的核心问题是：先explode外层events数组后，会分别对event_properties和entities两个数组执行explode操作，这会触发笛卡尔积——每个event_properties元素会和所有entities元素配对，生成大量重复行，无法实现按数组索引一一对应的需求。

解决方案 #

使用arrays_zip函数将event_properties和entities按索引打包成结构体数组，再执行一次explode，即可保证两个数组元素一一对应，避免重复。

修改后的完整代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, arrays_zip, col

# 初始化SparkSession
spark = SparkSession.builder \
    .master("local[1]") \
    .appName("PySpark Flatten JSON") \
    .getOrCreate()

# 读取JSON文件
df = spark.read.option("multiline", "true").json(r"C:\Users\Lajo\Downloads\spark_ex1_input.json")

# 第一步：展开events数组，拆分出单个event条目
df_events = df.select(explode(col("events")).alias("event"))

# 第二步：将两个数组按索引打包后展开，保证元素一一对应
df_flatten = df_events.select(
    col("event.event_name"),
    explode(arrays_zip(col("event.event_properties"), col("event.entities"))).alias("zipped"),
    col("event.event_timestamp")
).select(
    col("event_name"),
    col("zipped.event_properties"),
    col("zipped.entities"),
    col("event_timestamp")
)

# 查看结果
df_flatten.show()

代码说明 #

1. 展开外层数组：先用explode将events数组拆分为单独的event行，每行对应一个完整的event结构体。
2. 打包配对数组：arrays_zip会把event_properties和entities数组中同索引的元素打包成结构体，形成新的数组；再通过explode将这个结构体数组拆分为单独行，此时两个字段的元素完全一一对应。
3. 提取目标字段：最后从结构体和原event中提取需要的四个字段，得到符合预期的扁平化结果。

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