Spark join производит неправильные результаты

Question

Представляя здесь, прежде чем, возможно, зарегистрировать ошибку. Я использую Spark 1.6.0.

Это упрощенная версия проблемы, с которой я имею дело. Я отфильтровал таблицу, а затем попытаюсь сделать левое внешнее соединение с этим подмножеством и основной таблицей, соответствующей всем столбцам.

У меня есть только 2 строки в основной таблице и один в отфильтрованной таблице. Я ожидаю, что итоговая таблица будет иметь только одну строку из подмножества.

scala> val b = Seq(("a", "b", 1), ("a", "b", 2)).toDF("a", "b", "c") 
b: org.apache.spark.sql.DataFrame = [a: string, b: string, c: int] 

scala> val a = b.where("c = 1").withColumnRenamed("a", "filta").withColumnRenamed("b", "filtb") 
a: org.apache.spark.sql.DataFrame = [filta: string, filtb: string, c: int] 

scala> a.join(b, $"filta" <=> $"a" and $"filtb" <=> $"b" and a("c") <=> b("c"), "left_outer").show 
+-----+-----+---+---+---+---+ 
|filta|filtb| c| a| b| c| 
+-----+-----+---+---+---+---+ 
| a| b| 1| a| b| 1| 
| a| b| 1| a| b| 2| 
+-----+-----+---+---+---+---+ 

Я не ожидал такого результата. Я ожидал первую строку, но не вторую. Я подозревал, что это нулевое безопасное равенство, поэтому я попробовал это без него.

scala> a.join(b, $"filta" === $"a" and $"filtb" === $"b" and a("c") === b("c"), "left_outer").show 
16/03/21 12:50:00 WARN Column: Constructing trivially true equals predicate, 'c#18232 = c#18232'. Perhaps you need to use aliases. 
+-----+-----+---+---+---+---+ 
|filta|filtb| c| a| b| c| 
+-----+-----+---+---+---+---+ 
| a| b| 1| a| b| 1| 
+-----+-----+---+---+---+---+ 

ОК, это результат, которого я ожидал, но затем я стал подозревать предупреждение. Для решения этой проблемы существует отдельный вопрос StackOverflow: Spark SQL performing carthesian join instead of inner join

Поэтому я создаю новый столбец, который позволяет избежать предупреждения.

scala> a.withColumn("newc", $"c").join(b, $"filta" === $"a" and $"filtb" === $"b" and $"newc" === b("c"), "left_outer").show 
+-----+-----+---+----+---+---+---+ 
|filta|filtb| c|newc| a| b| c| 
+-----+-----+---+----+---+---+---+ 
| a| b| 1| 1| a| b| 1| 
| a| b| 1| 1| a| b| 2| 
+-----+-----+---+----+---+---+---+ 

Но теперь результат снова неправильный! У меня много нулевых проверок равенства, и предупреждение не является фатальным, поэтому я не вижу четкого пути к работе с ним/вокруг него.

Является ли поведение ошибкой, или это ожидаемое поведение? Если ожидается, почему?

Answer 1

Если вы хотите, ожидаемое использование поведения либо join об именах:

val b = Seq(("a", "b", 1), ("a", "b", 2)).toDF("a", "b", "c") 
val a = b.where("c = 1") 

a.join(b, Seq("a", "b", "c")).show 
// +---+---+---+ 
// | a| b| c| 
// +---+---+---+ 
// | a| b| 1| 
// +---+---+---+ 

или псевдонимы:

val aa = a.alias("a") 
val bb = b.alias("b") 

aa.join(bb, $"a.a" === $"b.a" && $"a.b" === $"b.b" && $"a.c" === $"b.c") 

Вы можете использовать <=>, а также:

aa.join(bb, $"a.a" <=> $"b.a" && $"a.b" <=> $"b.b" && $"a.c" <=> $"b.c") 

Насколько Я помню, что на какое-то время был особый случай для простого равенства. Вот почему вы получаете правильные результаты, несмотря на предупреждение.

Второе поведение выглядит действительно как ошибка, связанная с тем фактом, что у вас все еще есть a.c в ваших данных. Похоже, что он выбран ниже по течению до b.c, и оцениваемое условие на самом деле a.newc = a.c.

val expr = $"filta" === $"a" and $"filtb" === $"b" and $"newc" === $"c" 
a.withColumnRenamed("c", "newc").join(b, expr, "left_outer")