Блог

У меня есть два следующих фрейма данных, к которым я хочу присоединиться на основе col A

df1:

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

df2:

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

Поскольку столбец B одинаков, предположим, что существует логика выбора между ними, например, выберите

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

Простой способ добиться этого:

 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") 
 

Есть ли лучший способ добиться этого без переименования и удаления столбцов?

Это еще один метод использования selectExpr . Это экономит немного усилий при удалении столбцов.

 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()