如何在DataFrame中正确展开JSON并映射数组内对应对象？

你的核心问题是未按数组元素的位置索引进行匹配，分别explode两个数组会产生笛卡尔积，正确的做法是先将两个数组按位置配对再展开，或者通过位置索引关联。以下是两种高效实现方式：

方法一：使用arrays_zip+explode（推荐，简洁高效） #

该方法先将posts和subject数组按位置打包成结构体数组，再展开，直接保证对应位置的元素配对：

步骤（Spark DataFrame API） #

from pyspark.sql import functions as F

# 1. 展开外层的valueB数组（原valueB是长度为1的数组）
df = df.withColumn("valueB", F.explode(F.col("value B")))

# 2. 将posts和subject数组按位置打包成结构体数组
df = df.withColumn("post_subject_pair", F.arrays_zip(F.col("valueB.posts"), F.col("valueB.subject")))

# 3. 展开打包后的数组
df = df.withColumn("pair", F.explode(F.col("post_subject_pair")))

# 4. 提取目标字段并清理中间列
result_df = df.select(
    F.col("valueA"),
    F.col("pair.posts.body").alias("body"),
    F.col("pair.subject.id").alias("id"),
    F.col("pair.subject.name").alias("name"),
    F.col("pair.subject.timestamp").alias("timestamp"),
    F.col("valueB.communityName").alias("communityName")
)

# 查看结果
result_df.show(truncate=False)

对应Spark SQL实现 #

假设数据表名为my_table，注意列名含空格需用反引号包裹：

SELECT
    valueA,
    pair.posts.body AS body,
    pair.subject.id AS id,
    pair.subject.name AS name,
    pair.subject.timestamp AS timestamp,
    valueB.communityName AS communityName
FROM (
    SELECT
        valueA,
        explode(arrays_zip(valueB_exploded.posts, valueB_exploded.subject)) AS pair,
        valueB_exploded.communityName
    FROM (
        SELECT
            valueA,
            explode(`value B`) AS valueB_exploded
        FROM my_table
    ) t1
) t2

方法二：使用posexplode按索引关联 #

通过posexplode同时获取数组元素的位置索引，再通过索引关联两个展开后的数据集：

from pyspark.sql import functions as F

# 展开posts并保留位置索引
df_posts = df.withColumn("valueB", F.explode(F.col("value B"))) \
    .select(
        "valueA",
        "valueB.communityName",
        F.posexplode(F.col("valueB.posts")).alias("idx", "post")
    )

# 展开subject并保留位置索引
df_subjects = df.withColumn("valueB", F.explode(F.col("value B"))) \
    .select(
        "valueA",
        F.posexplode(F.col("valueB.subject")).alias("idx", "subject")
    )

# 通过valueA和索引idx关联，保证位置匹配
result_df = df_posts.join(df_subjects, on=["valueA", "idx"], how="inner") \
    .select(
        "valueA",
        F.col("post.body").alias("body"),
        F.col("subject.id").alias("id"),
        F.col("subject.name").alias("name"),
        F.col("subject.timestamp").alias("timestamp"),
        "communityName"
    )

最终期望输出 #

执行上述代码后，将得到如下结构的结果表：

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