PySpark中GroupedData不可下标访问报错及数据校验问题排查

错误原因 #

你遇到的TypeError: 'GroupedData' object is not subscriptable错误，是因为PySpark中groupBy()方法返回的是GroupedData对象，而非DataFrame，不能像操作DataFrame那样用['version']下标去指定字段。原代码里的df_2.groupBy(["date", "scope"])['version'].max()写法不符合PySpark的API规范，才触发了这个错误。

逻辑问题 #

你的核心逻辑存在两处漏洞：

• 分组取最大version后，只保留了date、scope和最大version值，丢失了对应的my_id信息，后续无法通过my_id和df_1关联校验。
• 没有从df_2中筛选出每个scope+date组合下对应最大version的所有行，直接关联的话无法定位到最新版本的完整数据。

修正后的代码 #

以下是符合需求的PySpark实现代码：

from pyspark.sql import functions as F

# 步骤1：获取每个scope+date对应的最大version
max_version_df = df_2.groupBy(["date", "scope"]) \
                     .agg(F.max("version").alias("max_version"))

# 步骤2：关联回原df_2，筛选出最新版本的所有记录
latest_version_records = df_2.join(
    max_version_df,
    on=["date", "scope", "version"],
    how="inner"
)

# 步骤3：用leftanti关联找出df_1中缺失的my_id（即最新版本存在但df_1没有的记录）
missing_records = latest_version_records.join(
    df_1,
    on=["my_id"],
    how="leftanti"
)

# 查看结果：这些是df_2最新版本里有，但df_1缺失的my_id及对应信息
missing_records.show()

代码说明 #

• 第一步通过groupBy+agg正确获取每个分组的最大version，并给结果字段起别名方便后续关联。
• 第二步将分组结果和原df_2关联，筛选出所有属于最新版本的记录，保留完整的my_id等信息。
• 第三步用leftanti关联，得到df_2最新版本中存在但df_1没有的记录，即需要补充的缺失数据。

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