Spark & ​​Scala: как заменить значения в Dataframes в разных столбцах

Question

У меня есть эти dataFrames:

+----+-------+-----------+...+------+----------------+---------+
|mot1|  brand|     device|...|action|Column_to_modify|New_value|
+----+-------+-----------+...------+----------------+---------+
|  09|  Tesla|         PC|...|modify|           brand|     Jeep|
|  10|  Tesla|SmallTablet|...|modify|           brand|     Jeep|
|  09|  Tesla|         PC|...|modify|           brand|     Jeep|
|  10|  Tesla|SmallTablet|...|modify|            mot1|       20|
|  10|  Tesla|SmallTablet|...|modify|            mot1|       20|
+----+-------+-----------+...+------+----------------+---------+

Так как я могу изменить столбцы, используя столбцы "Column_to_modify" и "New_value"?

Что я хочу:

+----+-------+-----------+...+------+----------------+---------+
|mot1|  brand|     device|...|action|Column_to_modify|New_value|
+----+-------+-----------+...------+----------------+---------+
|  09|   Jeep|         PC|...|modify|           brand|     Jeep|
|  10|   Jeep|SmallTablet|...|modify|           brand|     Jeep|
|  09|   Jeep|         PC|...|modify|           brand|     Jeep|
|  20|  Tesla|SmallTablet|...|modify|            mot1|       20|
|  20|  Tesla|SmallTablet|...|modify|            mot1|       20|
+----+-------+-----------+...+------+----------------+---------+

Есть идеи?

Answer 1

С UDF, назначенным для каждого столбца:

val df = List(
  ("09", "Tesla", "PC", "modify", "brand", "Jeep"),
  ("10", "Tesla", "SmallTablet", "modify", "brand", "Jeep"),
  ("09", "Tesla", "PC", "modify", "brand", "Jeep"),
  ("10", "Tesla", "SmallTablet", "modify", "mot1", "20"),
  ("10", "Tesla", "SmallTablet", "modify", "mot1", "20")
).toDF("mot1", "brand", "device", "action", "Column_to_modify", "New_value")

val modifyColumn = (colName: String, colValue: String, modifyColumnName: String, modifyColumnValue: String) =>
  if (colName.equals(modifyColumnName)) modifyColumnValue else colValue

val modifyColumnUDF = udf(modifyColumn)

val result = df
  .withColumn("mot1", modifyColumnUDF(lit("mot1"), $"mot1", $"Column_to_modify", $"New_value"))
  .withColumn("brand", modifyColumnUDF(lit("brand"), $"brand", $"Column_to_modify", $"New_value"))
result.show(false)

Выход:

+----+-----+-----------+------+----------------+---------+
|mot1|brand|device     |action|Column_to_modify|New_value|
+----+-----+-----------+------+----------------+---------+
|09  |Jeep |PC         |modify|brand           |Jeep     |
|10  |Jeep |SmallTablet|modify|brand           |Jeep     |
|09  |Jeep |PC         |modify|brand           |Jeep     |
|20  |Tesla|SmallTablet|modify|mot1            |20       |
|20  |Tesla|SmallTablet|modify|mot1            |20       |
+----+-----+-----------+------+----------------+---------+

Answer 2

Одним из быстрых способов достижения этого является использование операции map и преобразование в желаемый формат данных, как показано ниже:

import org.json.JSONObject

// creating input dataframe by reading input file
val inputDF = sparkSession.read.option("header", "true").csv("my_input_file.csv")
inputDF.printSchema()
inputDF.show(false)

val resultRDD = inputDF.toJSON.rdd.map(row => {
  val json = new JSONObject(row)

  val columnToModify = json.getString("Column_to_modify")
  val newValue = json.get("New_value")

  if (json.has(columnToModify)) {
    json.put(columnToModify, newValue)
  }

  json.toString
})

// converting the result RDD into dataframe
val finalOutputDF = sparkSession.read.json(resultRDD)
finalOutputDF.printSchema()
finalOutputDF.show(false)

, и результат будет следующим:

root
 |-- mot1: string (nullable = true)
 |-- brand: string (nullable = true)
 |-- device: string (nullable = true)
 |-- action: string (nullable = true)
 |-- Column_to_modify: string (nullable = true)
 |-- New_value: string (nullable = true)

+----+-----+-----------+------+----------------+---------+
|mot1|brand|device     |action|Column_to_modify|New_value|
+----+-----+-----------+------+----------------+---------+
|09  |Tesla|PC         |modify|brand           |Jeep     |
|10  |Tesla|SmallTablet|modify|brand           |Jeep     |
|09  |Tesla|PC         |modify|brand           |Jeep     |
|10  |Tesla|SmallTablet|modify|mot1            |20       |
|10  |Tesla|SmallTablet|modify|mot1            |20       |
+----+-----+-----------+------+----------------+---------+

root
 |-- Column_to_modify: string (nullable = true)
 |-- New_value: string (nullable = true)
 |-- action: string (nullable = true)
 |-- brand: string (nullable = true)
 |-- device: string (nullable = true)
 |-- mot1: string (nullable = true)

+----------------+---------+------+-----+-----------+----+
|Column_to_modify|New_value|action|brand|device     |mot1|
+----------------+---------+------+-----+-----------+----+
|brand           |Jeep     |modify|Jeep |PC         |09  |
|brand           |Jeep     |modify|Jeep |SmallTablet|10  |
|brand           |Jeep     |modify|Jeep |PC         |09  |
|mot1            |20       |modify|Tesla|SmallTablet|20  |
|mot1            |20       |modify|Tesla|SmallTablet|20  |
+----------------+---------+------+-----+-----------+----+

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