如何替换具有的数据框列？ spark scala中列平均值的符号？答案

【问题标题】：How to replace dataframe columns having ? symbol with mean value of the column in spark scala?如何替换具有的数据框列？ spark scala中列平均值的符号？
【发布时间】：2021-10-08 17:52:46
【问题描述】：

具有以下列的数据框：

one	two	three	four
3	?	Jaun	3.47
3	164	Jaun	3.47
1	?	?	2.68
3	164	Kaul	?
1	?	?	2.68

需要替换具有“？”的 df 列根据其数据类型的平均值和模式。如果该列是 Int 类型 -> 需要替换为均值如果列是String类型->需要用mode替换

预期输出：

one	two	three	four
3	65.6	Jaun	3.47
3	164	Jaun	3.47
1	65.6	Jaun	2.68
3	164	Kaul	2.46
1	65.6	Jaun	2.68

【问题讨论】：

什么意思？什么模式？
对应数据框列的平均值

标签： sql apache-spark replace mean mode

【解决方案1】：

以下代码将产生所需的结果，但可能需要一些优化。

  val spark = SparkSession.builder().master("local[*]").getOrCreate()
  import spark.implicits._
  spark.sparkContext.setLogLevel("ERROR")

  val inDF = // Read data

  inDF
    .withColumn("two", when('two === "?",
      mean(when('two === "?", 0).otherwise('two)).over()).otherwise('two))
    .withColumn("four", when('four === "?",
      mean(when('four === "?", 0).otherwise('four)).over()).otherwise('four))
    .withColumn("no_occurrence", count("*").over(Window.partitionBy("three")))
    .withColumn("max_occurrence", when('three =!= "?", max('no_occurrence).over()).otherwise(0))
    .withColumn("replacement", max(when('no_occurrence === 'max_occurrence, 'three)).over())
    .withColumn("three", when('three === "?", 'replacement).otherwise('three))
    .drop("no_occurrence", "max_occurrence", "replacement")
    .show(false)

    +---+----+-----+----+
    |one|two |three|four|
    +---+----+-----+----+
    |3  |65.6|Jaun |3.47|
    |3  |164 |Jaun |3.47|
    |3  |164 |Kaul |2.46|
    |1  |65.6|Jaun |2.68|
    |1  |65.6|Jaun |2.68|
    +---+----+-----+----+

【讨论】：

谢谢。但是为了获得正确的平均值，我们最初用 0 替换，然后用平均值改变那个 0。但这可能会导致已经存在的 0 值具有平均值，对吧？
修复了这个问题。已编辑帖子，请看一下。
谢谢莫哈娜。你能解释一下no_occurence、max_occurence和replacement部分吗
为了说明我们在 spark 中内置了功能，但在 mode 中没有，我使用 Windows 功能来实现这一点。 no_occurrences 列将有-一个特定单词在一列中重复多少次。 max_occurrence 列将具有 - no_occurrence 的最大计数。我将 0 标记为带有“？”的行的这 2 列的值。如果我们不标记零，那么？将被考虑进行计算。替换将有哪个字符串替换为？，将通过检查哪一行有 no_occurences==max_occurrence 来得到它。在每个 withColumn 之后，您可以调用 show() 方法并验证这些。
非常感谢 Mohana.. 用模式替换的另一种优化方法 inDF.withColumn("three",when(col("three") === "?",inDF.groupBy (col("three")).count().orderBy(desc("count")).first()(0)) .otherwise(col("three")))

【解决方案2】：

该模式有点棘手，因为 SparkSQL 不支持 mode() 或类似功能。但是你可以使用窗口函数：

select t.*,
       coalesce(two, two_avg),
       coalesce(three,
                max(case when three_cnt = max_three_cnt then three end) over ()
               ),
       coalesce(four, four_avg)
from (select t.*,
             max(three_cnt) over () as max_three_cnt
      from (select t.*,
                   avg(two) over () as two_avg,
                   count(*) over (partition by three) as three_cnt,
                   avg(four) over () as four_avg
            from t
           ) t
     ) t;

【讨论】：