如何在join关联操作中复用DataFrame，解决多次关联仅首次生效问题

核心原因 #

你遇到的问题由两个常见原因导致：

1. 未持久化的DataFrame每次被引用都会重新执行完整的计算链路，如果df1是经过多步计算得到的中间表，多次调用可能出现数据不一致，也会额外消耗性能。
2. 两次关联相同列名的df1会出现列名冲突，Spark无法区分两次关联进来的同名字段，导致第二次关联的数据读取歧义，看起来没有返回正常结果。

解决方案 #

• 关联时给同一DataFrame设置不同别名，或提前重命名列避免冲突
  两种实现方式可选：
  1. 关联时直接指定别名，后续取数也通过别名区分字段

  from pyspark.sql.functions import col
  
  df.join(df1.alias("df1_product"), df.product_type == col("df1_product.id"), "left")\
    .join(df1.alias("df1_deal"), df.deal_type == col("df1_deal.id"), "left")
  # 后续取值示例：col("df1_product.name") 为产品类型名称，col("df1_deal.name") 为交易类型名称
  

  2. 提前重命名df1的列生成两个无冲突的中间表再关联

  # 给关联产品类型的df1加统一前缀
  df1_product = df1.select([col(c).alias(f"product_{c}") for c in df1.columns])
  # 给关联交易类型的df1加统一前缀
  df1_deal = df1.select([col(c).alias(f"deal_{c}") for c in df1.columns])
  
  df.join(df1_product, df.product_type == df1_product.product_id, "left")\
    .join(df1_deal, df.deal_type == df1_deal.deal_id, "left")
  

• 对复用的DataFrame做持久化避免重复计算
  如果df1的计算链路较长，提前做缓存或持久化，既可以提升性能也能避免两次计算数据不一致的问题：

  # 内存充足时缓存到内存
  df1.cache()
  # 内存不足可选择落磁盘的存储级别
  # from pyspark import StorageLevel
  # df1.persist(StorageLevel.MEMORY_AND_DISK)
  
  # 所有逻辑运行完后可主动释放缓存
  # df1.unpersist()
  

注意：关联完成后不要直接使用原df1的列名取值，必须通过别名或者重命名后的字段名取值，否则依然会出现数据异常。

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