Как получить наиболее частое непустое значение в столбце?

Question

У меня есть следующий DataFrame df:

+-------------------+--------+--------------------+
|   id|         name|    type|                 url|
+-------------------+--------+--------------------+
|    1|      NT Note|    aaaa|                null|
|    1|      NT Note|    aaaa|http://www.teleab...|
|    1|      NT Note|    aaaa|http://www.teleab...|
|    1|      NT Note|    aaaa|                null|
|    1|      NT Note|    aaaa|                null|
|    2|          ABC|    bbbb|                null|
|    2|          ABC|    bbbb|                null|
|    2|          ABC|    bbbb|                null|
|    2|          ABC|    bbbb|                null|
+-------------------+--------+--------------------+

Я присваиваю наиболее частые значения url и type каждому узлу:

def windowSpec = Window.partitionBy("id", "url", "type") 
val result = df.withColumn("count", count("url").over(windowSpec))  
  .orderBy($"count".desc)                                                                                 
  .groupBy("id")                                                                                     
  .agg(
  first("url").as("URL"),
  first("type").as("Typel")
)

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

+-------------------+--------+--------------------+
|   id|         name|    type|                 url|
+-------------------+--------+--------------------+
|    1|      NT Note|    aaaa|http://www.teleab...|
|    2|          ABC|    bbbb|                null|
+-------------------+--------+--------------------+

Теперь я получаю вывод, показанный ниже, потому что null чаще для идентификатора записи 1:

+-------------------+--------+--------------------+
|   id|         name|    type|                 url|
+-------------------+--------+--------------------+
|    1|      NT Note|    aaaa|                null|
|    2|          ABC|    bbbb|                null|
+-------------------+--------+--------------------+

Answer 1

Вы можете сделать это, используя udf, как показано ниже

import org.apache.spark.sql.functions._
import scala.collection.mutable.WrappedArray

//function to return most frequent url

def mfnURL(arr: WrappedArray[String]): String = {
        val filterArr = arr.filterNot(_ == null)
        if (filterArr.length == 0)
            return null
        else {
            filterArr.groupBy(identity).maxBy(_._2.size)._1
        }
    }

//registering udf mfnURL

val mfnURLUDF = udf(mfnURL _)

//applying groupby , agg and udf

df.groupBy("id", "name", "type").agg(mfnURLUDF(collect_list("url")).alias("url")).show

//Sample output

+---+-------+----+--------------------+
| id|   name|type|                 url|
+---+-------+----+--------------------+
|  2|    ABC|bbbb|                null|
|  1|NT Note|aaaa|http://www.teleab...|
+---+-------+----+--------------------+