Spark 中的累积积

Question

我尝试在 Spark Scala 中实现一个累积积，但我真的不知道该怎么做。我有以下数据框：

Input data:
+--+--+--------+----+
|A |B | date   | val|
+--+--+--------+----+
|rr|gg|20171103| 2  |
|hh|jj|20171103| 3  |
|rr|gg|20171104| 4  |
|hh|jj|20171104| 5  |
|rr|gg|20171105| 6  |
|hh|jj|20171105| 7  |
+-------+------+----+

我想得到以下输出：

Output data:
+--+--+--------+-----+
|A |B | date   | val |
+--+--+--------+-----+
|rr|gg|20171105| 48  | // 2 * 4 * 6
|hh|jj|20171105| 105 | // 3 * 5 * 7
+-------+------+-----+

Answer 1

只要数字严格为正（如果存在，也可以使用 coalesce 处理 0），最简单的解决方案是计算对数之和并取指数:

import org.apache.spark.sql.functions.{exp, log, max, sum}

val df = Seq(
  ("rr", "gg", "20171103", 2), ("hh", "jj", "20171103", 3), 
  ("rr", "gg", "20171104", 4), ("hh", "jj", "20171104", 5), 
  ("rr", "gg", "20171105", 6), ("hh", "jj", "20171105", 7)
).toDF("A", "B", "date", "val")

val result = df
  .groupBy("A", "B")
  .agg(
    max($"date").as("date"), 
    exp(sum(log($"val"))).as("val"))

由于这使用了 FP 算法，因此结果将不准确：

result.show

+---+---+--------+------------------+
|  A|  B|    date|               val|
+---+---+--------+------------------+
| hh| jj|20171105|104.99999999999997|
| rr| gg|20171105|47.999999999999986|
+---+---+--------+------------------+

但四舍五入后应该足以满足大多数应用程序的需求。

result.withColumn("val", round($"val")).show

+---+---+--------+-----+
|  A|  B|    date|  val|
+---+---+--------+-----+
| hh| jj|20171105|105.0|
| rr| gg|20171105| 48.0|
+---+---+--------+-----+

如果这还不够，您可以定义 UserDefinedAggregateFunction 或 Aggregator () 或使用功能 API 和 reduceGroups:

import scala.math.Ordering

case class Record(A: String, B: String, date: String, value: Long)

df.withColumnRenamed("val", "value").as[Record]
  .groupByKey(x => (x.A, x.B))
  .reduceGroups((x, y) => x.copy(
    date = Ordering[String].max(x.date, y.date),
    value = x.value * y.value))
  .toDF("key", "value")
  .select($"value.*")
  .show

+---+---+--------+-----+
|  A|  B|    date|value|
+---+---+--------+-----+
| hh| jj|20171105|  105|
| rr| gg|20171105|   48|
+---+---+--------+-----+

Answer 2

您可以使用 collect_list+UDF 或 UDAF 解决此问题。 UDAF可能更高效，但由于本地聚合更难实现。

如果您有这样的数据框：

+---+---+
|key|val|
+---+---+
|  a|  1|
|  a|  2|
|  a|  3|
|  b|  4|
|  b|  5|
+---+---+

您可以调用 UDF :

val prod = udf((vals:Seq[Int]) => vals.reduce(_ * _))

df
  .groupBy($"key")
  .agg(prod(collect_list($"val")).as("val"))
  .show()

+---+---+
|key|val|
+---+---+
|  b| 20|
|  a|  6|
+---+---+

Answer 3

从 Spark 2.4 开始，您还可以使用高阶函数 aggregate:

来计算它

import org.apache.spark.sql.functions.{expr, max}
val df = Seq(
  ("rr", "gg", "20171103", 2),
  ("hh", "jj", "20171103", 3),
  ("rr", "gg", "20171104", 4),
  ("hh", "jj", "20171104", 5),
  ("rr", "gg", "20171105", 6),
  ("hh", "jj", "20171105", 7)
).toDF("A", "B", "date", "val")

val result = df
  .groupBy("A", "B")
  .agg(
    max($"date").as("date"),
    expr("""
   aggregate(
     collect_list(val),
     cast(1 as bigint),
     (acc, x) -> acc * x)""").alias("val")
  )

Answer 4

Spark 3.2+

product(e: Column): Column
Aggregate function: returns the product of all numerical elements in a group.

Scala

import spark.implicits._
var df = Seq(
    ("rr", "gg", 20171103, 2),
    ("hh", "jj", 20171103, 3),
    ("rr", "gg", 20171104, 4),
    ("hh", "jj", 20171104, 5),
    ("rr", "gg", 20171105, 6),
    ("hh", "jj", 20171105, 7)
).toDF("A", "B", "date", "val")

df = df.groupBy("A", "B").agg(max($"date").as("date"), product($"val").as("val"))
df.show(false)
// +---+---+--------+-----+
// |A  |B  |date    |val  |
// +---+---+--------+-----+
// |hh |jj |20171105|105.0|
// |rr |gg |20171105|48.0 |
// +---+---+--------+-----+

PySpark

from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.getOrCreate()
data = [('rr', 'gg', 20171103, 2),
        ('hh', 'jj', 20171103, 3),
        ('rr', 'gg', 20171104, 4),
        ('hh', 'jj', 20171104, 5),
        ('rr', 'gg', 20171105, 6),
        ('hh', 'jj', 20171105, 7)]
df = spark.createDataFrame(data, ['A', 'B', 'date', 'val'])

df = df.groupBy('A', 'B').agg(F.max('date').alias('date'), F.product('val').alias('val'))
df.show()
#+---+---+--------+-----+
#|  A|  B|    date|  val|
#+---+---+--------+-----+
#| hh| jj|20171105|105.0|
#| rr| gg|20171105| 48.0|
#+---+---+--------+-----+

Spark 中的累积积

Cumulative product in Spark

scala

product

aggregation

apache-spark

apache-spark-sql