集計

初期状態と配列内のすべての要素にバイナリ演算子を適用し、これを単一の状態に縮小します。最終状態は、終了関数を適用することによって最終結果に変換されます。Spark Connect をサポートします。

対応する Databricks SQL 関数については、 aggregate関数を参照してください。

構文

Python

from pyspark.sql import functions as dbf

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

パラメーター

パラメーター	Type	説明
col	pyspark.sql.Column または str	列または式の名前。
initialValue	pyspark.sql.Column または str	初期値。列または式の名前。
merge	function	initialValue と同じ型の式を返すバイナリ関数。
finish	functionオプション	累積値を変換するために使用されるオプションの単項関数。

パラメーター	Type	説明
col	pyspark.sql.Column または str	列または式の名前。
initialValue	pyspark.sql.Column または str	初期値。列または式の名前。
merge	function	initialValue と同じ型の式を返すバイナリ関数。
finish	functionオプション	累積値を変換するために使用されるオプションの単項関数。

戻り値

pyspark.sql.Column: 集計関数を適用した後の最終値。

例

例1 : 合計を使った単純な集計

Python

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()

Output

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

例2 : 終了関数を使った集計

Python

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()

Output

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