集計

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

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

構文

Python

from pyspark.databricks.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オプション	累積値を変換するために使用されるオプションの単項関数。

戻り値

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

例

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

Python

from pyspark.databricks.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.databricks.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|
+----+