请求将PySpark SQL多条件关联转换为DataFrame API关联

你遇到的这个报错是因为CDH5.10.2对应的Spark 1.6.x版本中，substring函数的长度参数只支持固定数值，不能传入Column对象——而你的代码里用了df1["j"]和df2["j"]+1这类Column作为长度参数，直接触发了类型错误。

下面是完整的、修复后的DataFrame API写法，完美对应你原来的SQL逻辑：

步骤1：导入必要的函数 #

from pyspark.sql import functions as F

步骤2：构建关联条件并执行join #

这里我们用F.expr()来直接复用SQL风格的表达式，绕开旧版本substring的限制，同时给DataFrame起别名让字段引用更清晰：

# 定义关联条件
join_conditions = [
    F.expr("concat(substr(upper(trim(d1.a)), 0, d1.j), ' ') = substr(upper(trim(d2.j)), 0, d2.j + 1)"),
    F.upper(F.trim(df1["c"])) == F.upper(F.trim(df2["f"]))
]

# 执行内关联（和原SQL的join一致）
join_df = df1.alias("d1").join(df2.alias("d2"), on=join_conditions, how="inner")

步骤3：完成字段选择、过滤和常量列添加 #

这一步完全对应原SQL的select和where逻辑，同时注意PySpark中多条件过滤要用&连接，且每个条件需加括号避免优先级问题：

final_df = join_df.select(
    F.col("d1.a"),
    F.col("d1.b"),
    F.col("d1.c").alias("aaa"),
    F.col("d2.d"),
    F.col("d2.e"),
    F.col("d2.f"),
    F.col("d2.g"),
    F.col("d2.h"),
    F.col("d2.i"),
    F.col("d2.j").alias("length"),
    F.lit(month_end).alias("month_end")  # 对应原SQL的'{1}' as month_end
).where(
    (F.length(F.upper(F.trim(F.col("d2.i")))) > F.col("d2.j")) &
    (F.length(F.upper(F.trim(F.col("d1.a")))) == (F.col("d1.j") + 3))
)

这样写既完全复现了你原SQL的业务逻辑，又解决了旧版本Spark中substring函数的参数限制问题，运行起来就不会报错了。

内容的提问来源于stack exchange，提问作者A.N.Gupta