Как обновить Spark DataFrame на основе SQL-Update-Query с помощью Inner-Join

Question

У меня есть SQL-запрос, который обновляет table как

Update table
set
  a.Column1 = b.Column1,
  a.Column2 = b.Column2,
  a.Column3 = b.Column3,
  .....
From table a
Inner Join (
  Select 
    ColumnA,
    min(Column1),
    max(Column2),
    max(Column3),
    .....
  From table
  Group By ColumnA
) b
On a.ColumnB = b.ColumnA

. Я хочу создать ту же функциональность с помощью Spark-DataFrames, что означает, что я хочу обновить DataFrame table_df, имеющую столбцыкак table в операторе SQL-Update.Подзапрос в Inner Join возвращает одну строку для каждого существующего значения в ColumnA из table.Это минимальное / максимальное значение является new_value для обновления, когда On -Statement of Join является True .

Я думаю, что это возможно при использовании .withColumnи when() как

table_df.withColumn("Column1", when(ColumnA == ColumnB,
           new_value).otherwise("Column1"))
        .withColumn("Column2", when(ColumnA == ColumnB, 
           new_value).otherwise("Column2"))
        ....

Но я действительно не знаю, как я могу гарантировать, что new_value будет совпадением min / max-значения таблицы, где ColumnA == ColumnB.Там также может быть более эффективное решение для этого.Кто-нибудь знает, как решить это эффективно?