在 Apache Spark Join 中包含空值

Question

我想在 Apache Spark 连接中包含空值。 Spark 默认不包含 null 的行。

这是默认的 Spark 行为。

 val numbersDf = Seq(
  ("123"),
  ("456"),
  (null),
  ("")
).toDF("numbers")

val lettersDf = Seq(
  ("123", "abc"),
  ("456", "def"),
  (null, "zzz"),
  ("", "hhh")
).toDF("numbers", "letters")

val joinedDf = numbersDf.join(lettersDf, Seq("numbers"))

这是 joinedDf.show() 的输出：

 +-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|       |    hhh|
+-------+-------+

这是我想要的输出：

 +-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|       |    hhh|
|   null|    zzz|
+-------+-------+

原文由 Powers 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

Spark 提供了一个特殊的 NULL 安全相等运算符：

 numbersDf
  .join(lettersDf, numbersDf("numbers") <=> lettersDf("numbers"))
  .drop(lettersDf("numbers"))

 +-------+-------+
|numbers|letters|
+-------+-------+
|    123|    abc|
|    456|    def|
|   null|    zzz|
|       |    hhh|
+-------+-------+

注意不要将它与 Spark 1.5 或更早版本一起使用。在 Spark 1.6 之前，它需要笛卡尔积（ SPARK-11111 - _快速空安全连接_）。

在 Spark 2.3.0 或更高版本中，您可以在 PySpark 中使用 Column.eqNullSafe ：

 numbers_df = sc.parallelize([
    ("123", ), ("456", ), (None, ), ("", )
]).toDF(["numbers"])

letters_df = sc.parallelize([
    ("123", "abc"), ("456", "def"), (None, "zzz"), ("", "hhh")
]).toDF(["numbers", "letters"])

numbers_df.join(letters_df, numbers_df.numbers.eqNullSafe(letters_df.numbers))

 +-------+-------+-------+
|numbers|numbers|letters|
+-------+-------+-------+
|    456|    456|    def|
|   null|   null|    zzz|
|       |       |    hhh|
|    123|    123|    abc|
+-------+-------+-------+

和 %<=>% 在 SparkR ：

 numbers_df <- createDataFrame(data.frame(numbers = c("123", "456", NA, "")))
letters_df <- createDataFrame(data.frame(
  numbers = c("123", "456", NA, ""),
  letters = c("abc", "def", "zzz", "hhh")
))

head(join(numbers_df, letters_df, numbers_df$numbers %<=>% letters_df$numbers))

   numbers numbers letters
1     456     456     def
2    <NA>    <NA>     zzz
3                     hhh
4     123     123     abc

使用 SQL ( Spark 2.2.0+ )，您可以使用 IS NOT DISTINCT FROM ：

 SELECT * FROM numbers JOIN letters
ON numbers.numbers IS NOT DISTINCT FROM letters.numbers

这也可以与 DataFrame API 一起使用：

 numbersDf.alias("numbers")
  .join(lettersDf.alias("letters"))
  .where("numbers.numbers IS NOT DISTINCT FROM letters.numbers")

原文由 zero323 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

基于 timothyzhang 的想法，可以通过删除重复列来进一步改进它：

 def dropDuplicateColumns(df: DataFrame, rightDf: DataFrame, cols: Seq[String]): DataFrame
= cols.foldLeft(df)((df, c) => df.drop(rightDf(c)))

 def joinTablesWithSafeNulls(rightDF: DataFrame, leftDF: DataFrame, columns: Seq[String], joinType: String): DataFrame =
{

val colExpr: Column = leftDF(columns.head) <=> rightDF(columns.head)

val fullExpr = columns.tail.foldLeft(colExpr) {
  (colExpr, p) => colExpr && leftDF(p) <=> rightDF(p)
}

val finalDF = leftDF.join(rightDF, fullExpr, joinType)

val filteredDF = dropDuplicateColumns(finalDF, rightDF, columns)

filteredDF

}

原文由 gavriil 发布，翻译遵循 CC BY-SA 4.0 许可协议