错误：连接中缺少已解决的属性

Question

我正在使用 pyspark 对两个表执行 join 的操作，这些表具有相对复杂的连接条件（在连接条件中使用大于/小于）。这工作正常，但只要我在加入前添加 fillna 命令就会崩溃。

代码如下所示：

join_cond = [
    df_a.col1 == df_b.colx,
    df_a.col2 == df_b.coly,
    df_a.col3 >= df_b.colz
]

df = (
    df_a
    .fillna('NA', subset=['col1'])
    .join(df_b, join_cond, 'left')
)

这会导致如下错误：

org.apache.spark.sql.AnalysisException: col1#6488,col2#4766,col3#4768,colx 中缺少已解决的属性 col1#4765 #4823,coly#4830,colz#4764 in operator !Join LeftOuter, (((col1#4765 = colx#4823) && (col2#4766 = coly#4830)) && (col3#4768 >= colz#4764))。具有相同名称的属性出现在操作中：col1。请检查是否使用了正确的属性。

在执行fillna 之后，spark 似乎不再识别col1。 （如果我将其注释掉，则不会出现错误。）问题是我确实需要该声明。 （总的来说，我已经大大简化了这个例子。）

我查看了this question，但这些答案对我不起作用。具体来说，在 fillna 之后使用 .alias('a') 不起作用，因为 spark 无法识别连接条件中的 a。

有人可以：

• 准确解释为什么会发生这种情况以及我以后如何避免这种情况？
• 请告诉我解决方法？

提前感谢您的帮助。

Answer 1

发生了什么？

为了“替换”空值，创建了一个包含新列的新数据框。这些新列与旧列具有相同的名称，但实际上是全新的 Spark 对象。在Scala code 中，您可以看到“更改”列是新创建的列，而原始列是dropped。

查看此效果的一种方法是在替换空值之前和之后在数据帧上调用explain：

df_a.explain()

打印

== Physical Plan ==
*(1) Project [_1#0L AS col1#6L, _2#1L AS col2#7L, _3#2L AS col3#8L]
+- *(1) Scan ExistingRDD[_1#0L,_2#1L,_3#2L]

同时

df_a.fillna(42, subset=['col1']).explain()

打印

== Physical Plan ==
*(1) Project [coalesce(_1#0L, 42) AS col1#27L, _2#1L AS col2#7L, _3#2L AS col3#8L]
+- *(1) Scan ExistingRDD[_1#0L,_2#1L,_3#2L]

两个计划都包含一个名为col1 的列，但在第一种情况下，内部表示称为col1#6L，而第二种情况称为col1#27L。

当连接条件df_a.col1 == df_b.colx 现在与列col1#6L 关联时，如果只有列col1#27L 是左表的一部分，则连接将失败。

如何解决问题？

显而易见的方法是在连接条件定义之前移动 `fillna` 操作：

df_a = df_a.fillna('NA', subset=['col1'])
join_cond = [
    df_a.col1 == df_b.colx,
[...]

如果这是不可能的或不希望的，您可以更改连接条件。除了使用数据框 (df_a.col1) 中的列之外，您还可以使用 col 函数使用与任何数据框不关联的列。此列仅根据其名称起作用，因此在数据框中替换该列时会忽略：

from pyspark.sql import functions as F
join_cond = [
    F.col("col1") == df_b.colx,
    df_a.col2 == df_b.coly,
    df_a.col3 >= df_b.colz
]

第二种方法的缺点是两个表中的列名必须是唯一的。