Spark执行inner join时报LeftSemi连接用PythonUDF不支持错误如何解决

• 你的猜测部分成立：空df1是触发本次报错的直接诱因，但不是根本原因。Spark的Catalyst优化器在检测到以下特征时，会自动把你写的inner join优化为LeftSemi Join来减少不必要的shuffle和计算：
  1. 左表（df1）仅包含join关联列customerHash且已经去重
  2. 内连接的结果本质上等价于取右表（df2）中关联列匹配左表的行
     空df1的场景会命中这个优化分支，所以虽然你没写LeftSemi，实际执行的是LeftSemi Join。
• 你没有显式定义Python UDF不代表执行链路中没有：大概率是df1或者df2的customerHash列在之前的加工流程中，用到了隐式Python UDF（比如pandas UDF、自定义Python UDF生成该列、或者该列的隐式类型转换用到了Python UDF），LeftSemi Join的关联条件不支持Python UDF，因此触发了报错。

按优先级从高到低排序：

1. 替换加工链路中的Python UDF（最优）
   排查customerHash列的生成逻辑，将用到的Python UDF全部替换为Spark原生SQL函数实现，完全消除Python UDF依赖。改造完成后可以直接用更符合你需求的LeftSemi Join写法，性能比inner join更好：

from pyspark.sql import functions as F

result = df2.join(
    df1.select("customerHash").distinct(),
    on="customerHash",
    how="left_semi"
)

2. 关闭自动转换LeftSemi Join的优化规则
   如果暂时没法改造Python UDF，可以在Spark会话级别关闭对应优化规则，避免触发异常：

# 关闭inner join自动转semi join的优化
spark.conf.set("spark.sql.optimizer.convertInnerToSemiJoin.enabled", False)

# 原有写法可正常执行
result = df1\
.select("customerHash")\
.distinct()\
.join(df2, ["customerHash"], 'inner')

3. 临时规避方案
   如果没有权限修改Spark配置，可以给df1加一个恒真的原生过滤条件，骗过优化器使其不触发自动转换：

result = df1\
.select("customerHash")\
.distinct()\
.filter(F.lit(1) == F.lit(1))\
.join(df2, ["customerHash"], 'inner')

注意：不要用isin加collect的写法，该写法会将df1的全量数据拉取到Driver节点，数据量稍大就会导致Driver OOM，且完全无法利用分布式计算能力，性能极低。

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