scala - 将列的元素除以按另一列元素分组的(同一列的)元素总和

Question

我一直在研究 aSspark 应用程序，并试图转换数据帧，如表 1 所示。我想将列 (_2) 的每个元素除以按另一列 (_1) 的元素分组的元素总和(同一列) )。表 2 是预期的结果。

表 1

+---+---+
| _1| _2|
+---+---+
|  0| 13|
|  0|  7|
|  0|  3|
|  0|  1|
|  0|  1|
|  1|  4|
|  1|  8|
|  1| 18|
|  1|  4|
+---+---+

表 2

+---+----+
| _1| _2 |
+---+----+
|  0|13/x|
|  0| 7/x|
|  0| 3/x|
|  0| 1/x|
|  0| 1/x|
|  1| 4/y|
|  1| 8/y|
|  1|18/y|
|  1| 4/y|
+---+----+

其中，x= (13+7+3+1+1) 和 y = (4+8+18+4)

然后，我想计算 _1 列中每个元素的熵:
即对于列 _1 中的每个元素计算 总和(p_i x log(p_i)) 在_2列中。其中，p_i 基本上是 中第 _1 列中每个值的第 _2 列中的值。表 2 .

最终输出将是。

+---+---------+
| _1| ENTROPY |
+---+---------+
|  0|entropy_1|
|  1|entropy_2|
+---+---------+

我怎样才能在 spark(最好在 scala)中实现这个？执行上述操作的优化方法是什么？我是 Scala 的新手，任何相关的建议将不胜感激。

谢谢你。

Answer 1

如果你想要一个简洁的解决方案并且组相当小，你可以使用窗口函数。首先你必须定义一个窗口:

import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy("_1").rowsBetween(Long.MinValue, Long.MaxValue)

可能性:

import org.apache.spark.sql.functions.sum

val p = $"_2" / sum($"_2").over(w)
val withP = df.withColumn("p", p)

最后是熵:

import org.apache.spark.sql.functions.log2

withP.groupBy($"_1").agg((-sum($"p" * log2($"p"))).alias("entropy"))

对于示例数据

val df = Seq(
  (0, 13), (0, 7), (0, 3), (0, 1), (0, 1), (1, 4), (1, 8), (1, 18), (1, 4)).toDF

结果是:

+---+------------------+
| _1|           entropy|
+---+------------------+
|  1|1.7033848993102918|
|  0|1.7433726580786888|
+---+------------------+

如果窗口函数在性能方面 Not Acceptable ，您可以尝试聚合连接聚合:

df.groupBy($"_1").agg(sum("_2").alias("total"))
  .join(df, Seq("_1"), "inner")
  .withColumn("p", $"_2" / $"total")
  .groupBy($"_1").agg((-sum($"p" * log2($"p"))).alias("entropy"))

在哪里:

df.groupBy($"_1").agg(sum("_2").alias("total"))

计算 _2 的总和来自 _1 ,

_.join(df, Seq("_1"), "inner")

将聚合列添加到原始数据中，

_.withColumn("p", $"_2" / $"total")

计算概率，并且:

_.groupBy($"_1").agg((-sum($"p" * log2($"p"))).alias("entropy"))

聚合以获得熵。