aggregate

Применяет двоичный оператор к исходному состоянию и всем элементам в массиве и сокращает это до одного состояния. Окончательное состояние преобразуется в окончательный результат путем применения функции завершения. Поддерживает Spark Connect.

Для соответствующей функции Databricks SQL смотрите функцию aggregate.

Синтаксис

from pyspark.sql import functions as dbf

dbf.aggregate(col=<col>, initialValue=<initialValue>, merge=<merge>, finish=<finish>)

Параметры

Параметр	Тип	Description
col	pyspark.sql.Column или str	Имя столбца или выражения.
initialValue	pyspark.sql.Column или str	Начальное значение. Имя столбца или выражения.
merge	function	Двоичная функция, возвращающая выражение того же типа, что и initialValue.
finish	functionнеобязательный	Необязательная унарная функция, используемая для преобразования накопленных значений.

Возвраты

pyspark.sql.Column: окончательное значение после применения статистической функции.

Примеры

Пример 1. Простая агрегирование с суммой

from pyspark.sql import functions as dbf
df = spark.createDataFrame([(1, [20.0, 4.0, 2.0, 6.0, 10.0])], ("id", "values"))
df.select(dbf.aggregate("values", dbf.lit(0.0), lambda acc, x: acc + x).alias("sum")).show()

+----+
| sum|
+----+
|42.0|
+----+

Пример 2. Агрегирование с помощью функции завершения

from pyspark.sql import functions as dbf
df = spark.createDataFrame([(1, [20.0, 4.0, 2.0, 6.0, 10.0])], ("id", "values"))
def merge(acc, x):
    count = acc.count + 1
    sum = acc.sum + x
    return dbf.struct(count.alias("count"), sum.alias("sum"))
df.select(
    dbf.aggregate(
        "values",
        dbf.struct(dbf.lit(0).alias("count"), dbf.lit(0.0).alias("sum")),
        merge,
        lambda acc: acc.sum / acc.count,
    ).alias("mean")
).show()

+----+
|mean|
+----+
| 8.4|
+----+