Как избежать повторяющихся вызовов после nullSafeJoin scala spark

Question

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

  def nullSafeJoin(leftDF: DataFrame, rightDF: DataFrame, joinOnColumns: Seq[String]) = {

    val dataset1 = leftDF.alias("dataset1")
    val dataset2 = rightDF.alias("dataset2")

    val firstColumn = joinOnColumns.head
    val colExpression: Column = (col(s"dataset1.$firstColumn").eqNullSafe(col(s"dataset2.$firstColumn")))

    val fullExpr = joinOnColumns.tail.foldLeft(colExpression) {
      (colExpression, p) => colExpression && (col(s"dataset1.$p").eqNullSafe(col(s"dataset2.$p")))
    }
    dataset1.join(dataset2, fullExpr)
  }

Последний объединенный набор данных имеет повторяющиеся столбцы. Я попытался удалить столбцы, используя псевдоним, подобный следующему:

dataset1.join(dataset2, fullExpr).drop(s"dataset2.$firstColumn")

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

Я пытаюсь создать базовую кодовую базу c, поэтому не хочу передавать список столбцов, которые должны быть выбраны в функцию (В случае удаления мне придется просто отбросить список joinOnColumns мы перешли к функции)

Любые указатели на то, как решить эту проблему, были бы очень полезны. Спасибо!

Редактировать: (Пример данных)

leftDF :
+------------------+-----------+---------+---------+-------+
|                 A|          B|        C|        D| status|
+------------------+-----------+---------+---------+-------+
|             14567|         37|        1|     game|Enabled|
|             14567|       BASE|        1|      toy| Paused|
|             13478|       null|        5|     game|Enabled|
|              2001|       BASE|        1|     null| Paused|
|              null|         37|        1|     home|Enabled|
+------------------+-----------+---------+---------+-------+

rightDF :
+------------------+-----------+---------+
|                 A|          B|        C|
+------------------+-----------+---------+
|               140|         37|        1|
|               569|       BASE|        1|
|             13478|       null|        5|
|              2001|       BASE|        1|
|              null|         37|        1|
+------------------+-----------+---------+

Final Join (Required):
+------------------+-----------+---------+---------+-------+
|                 A|          B|        C|        D| status|
+------------------+-----------+---------+---------+-------+
|             13478|       null|        5|     game|Enabled|
|              2001|       BASE|        1|     null| Paused|
|              null|         37|        1|     home|Enabled|
+------------------+-----------+---------+---------+-------+

Answer 1

Ваш окончательный DataFrame имеет повторяющиеся столбцы как из leftDF, так и из rightDF, не имеет идентификатора для проверки того, что этот столбец - из leftDF или rightDF.

Поэтому я переименовал столбцы leftDF и rightDF. Столбцы leftDF начинаются с left_[column_name], а столбцы rightDF начинаются с right_[column_name]

Надеюсь, приведенный ниже код поможет вам.

scala> :paste
// Entering paste mode (ctrl-D to finish)

  val left = Seq(("14567", "37", "1", "game", "Enabled"), ("14567", "BASE", "1", "toy", "Paused"), ("13478", "null", "5", "game", "Enabled"), ("2001", "BASE", "1", "null", "Paused"), ("null", "37", "1", "home", "Enabled")).toDF("a", "b", "c", "d", "status")
  val right = Seq(("140", "37", 1), ("569", "BASE", 1), ("13478", "null", 5), ("2001", "BASE", 1), ("null", "37", 1)).toDF("a", "b", "c")

  import org.apache.spark.sql.DataFrame
  def nullSafeJoin(leftDF: DataFrame, rightDF: DataFrame, joinOnColumns: Seq[String]):DataFrame = {
    val leftRenamedDF = leftDF
      .columns
      .map(c => (c, s"left_${c}"))
      .foldLeft(leftDF){ (df, c) =>
        df.withColumnRenamed(c._1, c._2)
      }
    val rightRenamedDF = rightDF
      .columns
      .map(c => (c, s"right_${c}"))
      .foldLeft(rightDF){(df, c) =>
        df.withColumnRenamed(c._1, c._2)
      }

    val fullExpr = joinOnColumns
      .tail
    .foldLeft($"left_${joinOnColumns.head}".eqNullSafe($"right_${joinOnColumns.head}")){(cee, p) =>
        cee && ($"left_${p}".eqNullSafe($"right_${p}"))
      }

    val finalColumns = joinOnColumns
      .map(c => col(s"left_${c}").as(c)) ++ // Taking All columns from Join columns
      leftDF.columns.diff(joinOnColumns).map(c => col(s"left_${c}").as(c)) ++ // Taking missing columns from leftDF
      rightDF.columns.diff(joinOnColumns).map(c => col(s"right_${c}").as(c)) // Taking missing columns from rightDF

    leftRenamedDF.join(rightRenamedDF, fullExpr).select(finalColumns: _*)
  }

scala>

Конечный результат DataFrame:

scala> nullSafeJoin(left, right, Seq("a", "b", "c")).show(false)


// Exiting paste mode, now interpreting.

+-----+----+---+----+-------+
|a    |b   |c  |d   |status |
+-----+----+---+----+-------+
|13478|null|5  |game|Enabled|
|2001 |BASE|1  |null|Paused |
|null |37  |1  |home|Enabled|
+-----+----+---+----+-------+