AWS Glue中PySpark多列关联重复及表合并需求咨询

嘿，我碰到过好几次这种AWS Glue里多表关联出重复数据的情况，给你梳理下解决思路和具体方案！

首先先明确你的核心需求：要把table_1和table_2基于相同字段关联，同时把table_2独有的两列加到结果里，table_1的旧数据在这两列填null，但多列关联时出现了重复数据——这个问题大概率是关联键不唯一或者连接方式/处理逻辑没做对导致的，下面分步骤解决：

第一步：先清理原始表的重复行 #

重复数据的根源很多时候是原始表里本身就有重复的关联键组合，比如id+code这种组合在table_1或table_2里有重复行，关联后就会产生笛卡尔积，导致结果翻倍。

先对两张表的关联键去重：

# 假设你的关联键是id、code、type这几个相同字段，根据实际情况调整
table_1_dedup = table_1.dropDuplicates(['id', 'code', 'type'])
table_2_dedup = table_2.dropDuplicates(['id', 'code', 'type'])

第二步：选择更合适的合并方式（二选一） #

方案一：用UnionByName替代Join（推荐，适合大部分列相同的场景） #

因为两张表Schema几乎一致，只是table_2多两列，这种情况用unionByName比Join更简洁，还能避免Join带来的重复问题：

1. 先给table_1补上table_2独有的两列，填充null（注意要和table_2的字段类型一致）
2. 直接合并两张表

代码示例：

from pyspark.sql.functions import lit

# 给table_1添加table_2独有的字段，类型和table_2保持一致
extra_col1_type = table_2.select('extra_col1').dtypes[0][1]
extra_col2_type = table_2.select('extra_col2').dtypes[0][1]

table_1_with_extra = table_1_dedup \
    .withColumn('extra_col1', lit(None).cast(extra_col1_type)) \
    .withColumn('extra_col2', lit(None).cast(extra_col2_type))

# 合并两张表，自动匹配列名，允许缺失列（这里table_1已经补了，其实也可以不用，但加上更稳妥）
merged_df = table_1_with_extra.unionByName(table_2_dedup, allowMissingColumns=True)

# 最后再做一次去重，确保结果干净
final_result = merged_df.dropDuplicates(['id', 'code', 'type'])

方案二：用Full Outer Join并严格控制关联条件 #

如果你必须用Join来实现（比如有特殊的匹配逻辑），那要注意用full_outer连接，并且明确所有关联字段，同时处理列名冲突：

from pyspark.sql.functions import coalesce

# 给表加别名，避免列名冲突
t1 = table_1_dedup.alias('t1')
t2 = table_2_dedup.alias('t2')

# 定义完整的关联条件，所有相同字段都要写上，不能漏
join_condition = (t1.id == t2.id) & \
                 (t1.code == t2.code) & \
                 (t1.type == t2.type)

# 执行全外连接，然后选择需要的列
result_df = t1.join(t2, join_condition, how='full_outer') \
    .select(
        # 共同字段用coalesce取非null值，确保不管是t1还是t2的数据都能保留
        coalesce(t1.id, t2.id).alias('id'),
        coalesce(t1.code, t2.code).alias('code'),
        coalesce(t1.type, t2.type).alias('type'),
        # 其他共同字段同理...
        # table_2独有的字段，t1没有的行自动是null
        t2.extra_col1.alias('extra_col1'),
        t2.extra_col2.alias('extra_col2')
    )

# 最后去重
final_result = result_df.dropDuplicates(['id', 'code', 'type'])

关键注意点 #

• 关联键必须完整：多列关联时，一定要把所有用来匹配的相同字段都写到关联条件里，少一个都可能导致部分匹配，产生重复行
• 去重要彻底：原始表和结果集都要做去重，尤其是关联键组合的去重
• 优先用Union：当两张表Schema高度一致时，Union比Join更高效，也更不容易出重复问题

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