ウィンドウ

ウィンドウ関数を使うと、集計に関与しない列を失うことなく、データフレームの行に集計値やその他の値を追加できます。

活用例

customer が行った各トランザクションを含む transactions データセットがあり、顧客ごとに average_spend を計算して各トランザクションに追加したいとします。結合を使ってこれを行うには、グループ化して平均を求めてから、元のテーブルに結合する必要があります。

Copied!
1
2
3
4
5
6
averages = transactions\
	.groupBy('customer')\
	.agg(
		F.avg('spend').alias('average_spend')
	)
transactions = transactions.join(averages, ['customer'], 'left_outer')

最大支出額を求めたい場合は、平均ではなく最大値を計算し、その最大値を使って元のテーブルに結合する必要があるため、ロジックはさらに複雑になります。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
maximums = transactions\
	.groupBy('customer')\
	.max(
		F.avg('spend').alias('max_spend')
	)
transactions = transactions\
	.join(
		averages,
		(transactions.customer == maximums.customer) &\
		(transactions.spend == maximums.max_spend),
		'left_outer'
	).drop(maximums.customer)

一方、ウィンドウ関数を使うと、まずウィンドウを定義し、そのウィンドウに対して集計を計算することで、このコードを簡素化できます。

Copied!
1
2
3
4
5
6
7
8
9
from pyspark.sql.window import Window

window = Window()\
	.partitionBy('customer')\
	.orderBy('spend')

transactions = transactions\
	.withColumn('average_spend', F.avg('spend').over(window))
	.withColumn('max_spend', F.max('spend').over(window))

また、ウィンドウでのみ使用できる関数もいくつかあります。これらはウィンドウ関数と呼ばれ、次のセクションで説明します。

ウィンドウ関数

dense_rank()

lag(col, count=1, default=None)

lead(col, count=1, default=None)

ntile(n)

percent_rank()

rank()

row_number()

window(timeColumn, windowDuration, slideDuration=None, startTime=None)