python-2.7 - PySpark.sql.filter 未按预期执行

Question

执行以下代码时遇到问题:

from pyspark.sql import functions as F
from pyspark.sql import Row, HiveContext

hc = HiveContext()
rows1 = [Row(id1 = '2', id2 = '1', id3 = 'a'),
         Row(id1 = '3', id2 = '2', id3 = 'a'),
         Row(id1 = '4', id2 = '3', id3 = 'b')]
df1 = hc.createDataFrame(rows1)
df2 = df1.filter(F.col("id3")=="a")
df3 = df1.join(df2, df1.id2 == df2.id1, "inner")

当我运行上面的代码时，df3 是一个空的 DataFrame。然而:
如果我将代码更改为下面，它会给出正确的结果(2 行的 DataFrame):

from pyspark.sql import functions as F
from pyspark.sql import Row, HiveContext

hc = HiveContext()
rows1 = [Row(id1 = '2', id2 = '1', id3 = 'a'),
         Row(id1 = '3', id2 = '2', id3 = 'a'),
         Row(id1 = '4', id2 = '3', id3 = 'b')]
df1 = hc.createDataFrame(rows1)
rows2 = [Row(id1 = '2', id2 = '1', id3 = 'a'),
         Row(id1 = '3', id2 = '2', id3 = 'a'),
         Row(id1 = '4', id2 = '3', id3 = 'b')]
df1_temp = hc.createDataFrame(rows2)
df2 = df1_temp.filter(F.col("id3")=="a")
df3 = df1.join(df2, df1.id2 == df2.id1, "inner")

所以我的问题是:为什么我必须在这里创建一个临时数据框？
另外，如果我无法在我的项目部分中获得 HiveContext，我该如何在现有数据帧之上创建一个重复的数据帧？

Answer 1

我相信您在这里遇到的问题是一个更普遍的问题的实例，其中某些类型的 DataFrame 自连接(包括 DataFrame 与自身过滤副本的连接)可能导致生成不明确或不正确的查询计划。

有几个与此相关的 Spark JIRA；以下是一些值得注意的:

SPARK-15063: "filtering and joining back doesn't work"似乎是您报告的特定类型实例的最接近匹配项。

SPARK-17154: "Wrong result can be returned or AnalysisException can be thrown after self-join or similar operations"对根本原因进行了很好的讨论。

还有其他 JIRA 票证处理这些问题的不同表现/方面。通过遵循从上面列出的票证开始的 JIRA“相关”链接链，可以发现这些票证。

这种歧义仅在通过 DataFrame 实例(通过下标，如 df["mycol"] 或通过字段访问，如 df.mycol )引用列时出现。可以通过给 DataFrame 取别名并通过别名引用列来避免这种歧义。例如，以下工作正常:

>>> from pyspark.sql import functions as F
>>> df1 = hc.createDataFrame(rows1).alias("df1")
>>> df2 = df1.filter(F.col("id3")=="a").alias("df2")
>>> df3 = df1.join(df2, F.col("df1.id2") == F.col("df2.id1"), "inner")
>>> df3.show()
+---+---+---+---+---+---+
|id1|id2|id3|id1|id2|id3|
+---+---+---+---+---+---+
|  4|  3|  b|  3|  2|  a|
|  3|  2|  a|  2|  1|  a|
+---+---+---+---+---+---+