python - 在 PySpark 中加入加入 None 值

Question

在 PySpark 中，我想使用键值对对两个 RDD 进行完全外连接，其中键可能为 None。例如:

rdd1 = sc.parallelize([(None, "a"), (None, "b")])
rdd2 = sc.parallelize([(None, "c"), (None, "d")])
join_rdd = rdd1.join(rdd2)

看起来 PySpark 连接了键为 None 的记录:

print(rdd1.join(rdd2).take(10))
>>> [(None, ('a', 'c')), (None, ('a', 'd')), (None, ('b', 'c')), (None, ('b', 'd'))]

但是，在 SQL 中，当我连接两个表时:

Table1:    Table2:
key   val   key   val
NULL  a     NULL  c
NULL  b     NULL  d

SELECT * FROM Table1 JOIN Table2 ON Table1.key = Table2.key

我有一个空结果集。

我认为这是因为在 Python 中 None == None 为 true，而在 SQL 中 NULL = NULL 为 false。

我有两个问题:

1. 有没有办法模拟 SQL 行为并强制 PySpark 不通过 None 加入？

2. 这是错误还是功能？作为 SQL 用户，我预计通过空键连接不会返回任何内容。我是 PySpark 的新手，在有关 joinig Nones 的文档中没有找到任何内容。也许值得在 Spark 编程指南中做一些说明？

还是我哪里错了？

谢谢!

Answer 1

你的期望是错误的。 RDD API 不遵循 SQL 语义，也从未打算这样做。 RDD.join 只是一个基于哈希的链接，带有 portable_hash其设计初衷是为了提供有意义的 None 哈希。

如果你想要类似 SQL 的语义，你应该使用 Spark SQL/Data Frames:

schema = StructType([
  StructField("_1", IntegerType(), True), StructField("_2", StringType(), False)
])

df1 = sqlContext.createDataFrame(rdd1, schema)
df2 = sqlContext.createDataFrame(rdd2, schema)
df1.join(df2, ["_1"])

如果您想在 RDD 上获得类似的结果，请在 join 之前过滤掉 None 键:

rdd1.filter(lambda x: x[0] is not None).join(rdd2)