集計とピボットテーブル

集計の構文

PySpark で集計を行う方法はいくつかあります。最も信頼性の高い方法として、この構文を推奨します。

Copied!
1
2
3
aggregated_df = df.groupBy('state').agg(
    F.max('city_population').alias('largest_city_in_state')
)

DataFrame の行をまとめずに、平均値や最大値などの集計値を各行に追加するには、ウィンドウ関数を使用します。ウィンドウ関数は元の列と行を保持します。

例

DataFrame の .groupBy() メソッドは、集計対象となる列を任意の数だけ受け取ります。また、DataFrame 全体を集計するには、列を指定しません。

Copied!
1
2
3
4
5
6
7
aggregated_df = df.groupBy('state', 'county').agg(
    F.max('city_population').alias('largest_city_in_state_county')
)

aggregated_df = df.groupBy().agg(
    F.max('city_population').alias('largest_city_overall')
)

グループ化された DataFrame の .agg() メソッドは、集計関数を任意の数だけ受け取ります。

Copied!
1
2
3
4
aggregated_df = df.groupBy('state').agg(
    F.max('city_population').alias('largest_city_in_state'),
    F.avg('city_population').alias('average_population_in_state')
)

デフォルトでは、集計によって aggregation_name(target_column) という形式の列が生成されます。ただし、Foundry の列名には、丸括弧やその他の英数字以外の文字を含めることはできません。代わりに、各集計に特定の名前をエイリアスとして設定します。

ピボットテーブル

PySpark のピボットテーブルは、通常のグループ化された集計とほぼ同じように動作します。

Copied!
1
pivoted_df = df.groupBy('equipment').pivot('sensor').mean('value')

集計関数

PySpark の集計関数について詳しくはこちら。↗

avg(column) / mean(column)

collect_list(column)

• すべての値を配列にまとめます。

collect_set(column)

• 重複を削除して、すべての値を配列にまとめます。

count(column)

corr(x, y)

• 列 x と y のピアソンの相関係数です。

covar_pop(col1, col2)

covar_samp(col1, col2)

countDistinct(column, *cols)

first(column, ignorenulls=False)

• グループ内の列の最初の値です。値が1つしか存在しないと想定されるものの、集計関数を選ぶ必要があるピボットテーブルで役立ちます。

grouping(column)

grouping_id(*cols)

kurtosis(column)

last(column, ignorenulls=False)

max(column)

min(column)

skewness(column)

stddev(column)

stddev_pop(column)

• 母集団の標準偏差です。

stddev_samp(column)

• 標本の不偏標準偏差です。

sum(column)

sumDistinct(column)

var_pop(column)

• 母分散です。

var_samp(column)

• 標本の不偏分散です。

variance(column)