Как заменить несколько строк, дублированных в одном столбце, одной строкой нулей в кадре данных Spark

Question

У меня есть кадр данных Spark, подобный этому:

+------+---------+---------+---------+---------+
| name | metric1 | metric2 | metric3 | metric4 | 
+------+---------+---------+---------+---------+
| a    |       1 |       2 |       3 |       4 | 
| b    |       1 |       2 |       3 |       4 | 
| c    |       3 |       1 |       5 |       4 | 
| a    |       3 |       3 |       3 |       3 | 
+------+---------+---------+---------+---------+

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

+------+---------+---------+---------+---------+
| name | metric1 | metric2 | metric3 | metric4 | 
+------+---------+---------+---------+---------+
| a    |    null |    null |    null |    null | 
| b    |       1 |       2 |       3 |       4 | 
| c    |       3 |       1 |       5 |       4 | 
+------+---------+---------+---------+---------+

Следующие работы:

import org.apache.spark.sql.functions._

val df = Seq(
  ("a", 1, 2, 3, 4), ("b", 1, 2, 3, 4), ("c", 3, 1, 5, 4), ("a", 3, 3, 3, 3)
).toDF("name", "metric1", "metric2", "metric3", "metric4")

val newDf = df
.groupBy(col("name"))
.agg(
  min(col("metric1")).as("metric1"),
  min(col("metric2")).as("metric2"),
  min(col("metric3")).as("metric3"),
  min(col("metric4")).as("metric4"),
  count(col("name")).as("NumRecords")
)
.withColumn("metric1", when(col("NumRecords") !== 1, lit(null)).otherwise(col("metric1")))
.withColumn("metric2", when(col("NumRecords") !== 1, lit(null)).otherwise(col("metric2")))
.withColumn("metric3", when(col("NumRecords") !== 1, lit(null)).otherwise(col("metric3")))
.withColumn("metric4", when(col("NumRecords") !== 1, lit(null)).otherwise(col("metric4")))
.drop("NumRecords")

но наверняка должен быть лучший путь ...

Answer 1

    scala> val df = Seq(("a", 1, 2, 3, 4), ("b", 1, 2, 3, 4), ("c", 3, 1, 5, 4), ("a", 3, 3, 3, 3)).toDF("name", "metric1", "metric2", "metric3", "metric4")
    scala> val newDf = df.groupBy(col("name")).agg(min(col("metric1")).as("metric1"),min(col("metric2")).as("metric2"),min(col("metric3")).as("metric3"),min(col("metric4")).as("metric4"),count(col("name")).as("NumRecords"))
    scala> val colArr2 = df.columns.diff(Array("name"))
    scala> val reqDF = colArr2.foldLeft(newDf){
     (df,colName)=>
     df.withColumn(colName,when(col("NumRecords") =!= "1",lit(null)).otherwise(col(colName)))
     }.drop("NumRecords")

    scala> reqDF.show
    +----+-------+-------+-------+-------+
    |name|metric1|metric2|metric3|metric4|
    +----+-------+-------+-------+-------+
    |   c|      3|      1|      5|      4|
    |   b|      1|      2|      3|      4|
    |   a|   null|   null|   null|   null|
    +----+-------+-------+-------+-------+

Пожалуйста, попробуйте, как указано выше.