Spark joins with condition on non join column

Question

I have two dataframes as follows which I want to join based on col A

df1:

+------+--------+-------+
|  A   |   B    |   C   |
+------+--------+-------+
| a1   |   5    |   asd |
| a2   |   12   |   asd |
+------+--------+-------+

df2:

+------+--------+-------+
|  A   |   B    |   D   |
+------+--------+-------+
|  a1  |   8    |   qwe |
|  a2  |   10   |   qwe |
+------+--------+-------+

Since column B is same lets assume there is a logic of choosing between the two for example choose the

+------+--------+------+-----+
|  A   |   B    |  C   |  D  |
+------+--------+------+-----+
|  a1  |   8    |  asd | qwe |
|  a2  |   12   |  asd | qwe |
+------+--------+-------+----+

A simple way to achieve this is:

val _df1 = df1.withColumnRenamed("B","B_df1")
val _df2 = df2.withColumnRenamed("B", "B_df2)
_df1.join(_df2, Seq("A"))
    .withColumn("B", when(col("B_df1") > col("B_df2"), 
 col("B_df1"))
    .otherwise(col("B_df2"))
    .drop(col("B_df1")
    .drop("B_df2")

Is there a better way to achieve this without renaming and dropping columns?

Shantanu Sharma Shantanu Sharma · Accepted Answer · 2019-04-17T20:00:31

This is another method using selectExpr. It saves little efforts of dropping columns.

import spark.implicits._

val df1 = Seq(("a1",5,"asd"),
              ("a2",12,"asd")
              ).toDF("A","B","C")

val df2 = Seq(("a1",8,"qwe"),
              ("a2",10,"qwe")
              ).toDF("A","B","D")


import org.apache.spark.sql.functions.col

df1.as("a").join(df2.as("b"), col("a.A") === col("b.A")).selectExpr("a.A AS A",
               "CASE WHEN a.B>b.B THEN a.B ELSE b.B END AS B",
               "a.C AS C",
               "b.D AS D").show()

Spark joins with condition on non join column

1 Answers