グループ処理#
グループごとの集計は、データ分析でよく使う操作です。売上を地域ごとに合計する、ログを時間ごとに数える、実験データを群ごとに平均する、どれも「まず行をグループに分け、次にグループごとに値を 1つ計算する」という同じ手順です。
Polars はこれを 2つの操作に分けます。group_by() がグループ分けを、agg() が計算を受け持ちます。2つの間には任意の式を置けるので、「各グループの先頭 3件」「各グループで最大値と平均を同時に」といった要求も、agg() の中の式を差し替えるだけで済みます。この章では、Polars のグループ処理を次の順に扱います。
group_by— 列の値でグループを作るagg— グループごとに集計するhaving— 集計結果でグループを絞るover— グループごとの計算結果を元の行に戻すgroup_by_dynamic/rolling— 時間軸や行番号での窓集計
どれも同じ「グループ化 → 集計」の枠組みの上にあり、結果を 1行にまとめるか、元の行に配るかだけが違います。
import polars as pl
from helper.jupyter import row
group_by#
DataFrame.group_by() はグループ化オブジェクト（GroupBy）を作ります。この時点ではまだ計算されません。agg() や mean() が呼ばれて初めて走ります。
ここで覚えておきたいのは、グループの順序が既定では保証されないという点です。並列に処理するため速い代わりに、結果の並びはデータの並びと一致しません。元のデータに現れた順で結果をたどりたいときは maintain_order=True を指定します。その代わり並列化の一部をあきらめることになり、速度は落ちます。
df = pl.DataFrame(
dict(
g=['A', 'B', 'C', 'A', 'B', 'B'],
x=[1, 2, 3, 4, 5, 6])
)
g = df.group_by('g', maintain_order=True)
type(g)
polars.dataframe.group_by.GroupBy
agg で集計#
GroupBy.agg() に式を渡すと、グループごとに計算して 1行にまとめます。よく使う集計は GroupBy のメソッド（.mean()、.first()、.sum() など）としても用意されており、これは agg() に同じ式を書いたのとまったく同じ結果になります。違うのは文字数だけです。複数の量を一度に計算するときは agg() が便利で、1回の呼び出しで複数の列を返せます。.mean() は 1列しか返せません。
row(
g.mean(),
g.agg(pl.all().mean()),
g.first(),
g.agg(pl.all().first())
)
shape: (3, 2)
|
shape: (3, 2)
|
shape: (3, 2)
|
shape: (3, 2)
|
agg() の中で複数の値を返す式（head(2) など）を使うと、結果の列はリストになります。agg() は 1グループにつき 1行しか残さないため、あふれた値をリストに入れるしかないからです。
GroupBy.head() と同じ形にしたいときは、explode() でリストを縦に展開します。展開後の行数は GroupBy.head() と一致します。2つの違いは、結果の行数を 1行に圧縮するかどうかだけです。
row(
g.head(2),
g.agg(pl.all().head(2)),
g.agg(pl.all().head(2)).explode('x')
)
shape: (5, 2)
|
shape: (3, 2)
|
shape: (5, 2)
|
.agg() の中で map_batches() を使ってユーザー関数を適用すると、関数が呼ばれる回数は行数ではなくグループ数になります。グループが多いほどこのコストは効いてきます。書き方と is_elementwise の影響は「GroupBy.agg() 内で map_batches() を使用する」 を参照してください。
having でグループを絞り込む#
集計したあとで「この条件を満たすグループだけ残す」という場面があります。注文件数が 10を超える顧客、平均点が 60未満のクラス、といった条件です。これは filter() では代用できません。条件に含まれるのがグループの統計量で、filter() は各行の値しか見えず、その行がどのグループに属するかを知らないからです。SQL の HAVING に相当するのが GroupBy.having() です。
引数には集約式を含む条件を渡します
戻り値は再び
GroupByで、続けてagg()やmean()を書けます
次の例では、グループごとの件数を数え、1件しかない C を落としてから平均を計算しています。
g.having(pl.col('g').count() > 1).mean()
| g | x |
|---|---|
| str | f64 |
| "A" | 2.5 |
| "B" | 4.333333 |
having() と filter() の違いは、対象が行かグループかという点です。having() はグループの統計量を見て、filter() はその行自身の値を見ます。したがって、having() に書く条件には必ず集約式が入り、filter() の条件には通常入りません。
メソッド |
絞り込む対象 |
|---|---|
|
行（row） |
|
グループ（group） |
over#
式の中で .over() を使うと、グループごとの計算結果を元の行に配れます。行数は変わりません。グループの統計量を各行の計算に使いたいときは over() が必要です。たとえば平均を引く、行の値をグループ内の最大値と比べる、といった処理です。
over() を使わず、集計してから結合しても同じ結果は得られます。ただし手順が増えます。まず 1グループ 1行の小さな表を作り、その表をグループのキーで元の表に結合し直す必要があります。結合のキーがユニークであることも自分で保証しなければなりません。次のコードは g ごとに x の平均を計算し、全行にその値を付けます。group_by().agg() が 1グループ 1行なのに対し、over() は元の行数を変えません。
df.with_columns(
pl.col('x').mean()
.over('g').alias('x_mean')
)
| g | x | x_mean |
|---|---|---|
| str | i64 | f64 |
| "A" | 1 | 2.5 |
| "B" | 2 | 4.333333 |
| "C" | 3 | 3.0 |
| "A" | 4 | 2.5 |
| "B" | 5 | 4.333333 |
| "B" | 6 | 4.333333 |
次のコードでは、g 列のグループごとに x 列の平均値を計算し、x 列から平均値を引いた結果を新しい列 x_sub_mean に格納します。グループごとの計算結果は元の行数と同じままで、対応する行にそのまま反映されます。
df.with_columns(
(pl.col('x') - pl.col('x').mean())
.over('g')
.alias('x_sub_mean')
)
| g | x | x_sub_mean |
|---|---|---|
| str | i64 | f64 |
| "A" | 1 | -1.5 |
| "B" | 2 | -2.333333 |
| "C" | 3 | 0.0 |
| "A" | 4 | 1.5 |
| "B" | 5 | 0.666667 |
| "B" | 6 | 1.666667 |
グループごとの結果の長さが 1でも元の行数と同じでもない場合、そのままではエラーになります。結果を行に戻す必要があるのに、長さが合わないと余った値をどの行に入れるか決まらないからです。mapping_strategy='join' を指定すると、結果をリストとして元の行に入れられます。次の例は、各グループの先頭 2件をリストにして新しい列に格納します。
df.with_columns(
pl.col('x').head(2)
.over('g', mapping_strategy='join')
.alias('x_head')
)
| g | x | x_head |
|---|---|---|
| str | i64 | list[i64] |
| "A" | 1 | [1, 4] |
| "B" | 2 | [2, 5] |
| "C" | 3 | [3] |
| "A" | 4 | [1, 4] |
| "B" | 5 | [2, 5] |
| "B" | 6 | [2, 5] |
次のコードでは、g 列のグループごとに x 列の最小値と最大値を求め、リストとして新しい列に格納します。
append() は演算式の中で 2つのスカラーを長さ 2のリストにまとめるので、mapping_strategy='join' と組み合わせるとちょうどよくはまります。
df.with_columns(
pl.col('x').min().append(pl.col('x').max())
.over('g', mapping_strategy='join')
.alias('x_min_max')
)
| g | x | x_min_max |
|---|---|---|
| str | i64 | list[i64] |
| "A" | 1 | [1, 4] |
| "B" | 2 | [2, 6] |
| "C" | 3 | [3, 3] |
| "A" | 4 | [1, 4] |
| "B" | 5 | [2, 6] |
| "B" | 6 | [2, 6] |
mapping_strategy='explode' にすると、各グループの結果を縦に並べます。これは group_by().agg().explode() と同じ結果で、間に物化を挟まない分だけ短く書けます。グループごとに複数行へ展開したいが列の構造は変えたくない、という場面に向いています。2つの戦略の違いは結果の行数だけです。join はリストを 1行に入れ、explode はリストを行に広げます。
df.select(
pl.col('g', 'x').head(2)
.over('g', mapping_strategy='explode')
)
| g | x |
|---|---|
| str | i64 |
| "A" | 1 |
| "A" | 4 |
| "B" | 2 |
| "B" | 5 |
| "C" | 3 |
filter() メソッドにも over() の式を使えます。条件は「この行が、属しているグループの統計量とどう関係するか」になります。over() なしで書くと、まずグループごとの平均を計算し、1グループ 1行の小さな表を作り、それをキーで元の表に結合し直してから絞る、という手順になります。次のコードでは、g 列のグループごとに、x 列の値がそのグループの平均値以上である行だけを絞り込んで出力します。
df.filter(
(pl.col('x') >= pl.col('x').mean())
.over('g')
)
| g | x |
|---|---|
| str | i64 |
| "C" | 3 |
| "A" | 4 |
| "B" | 5 |
| "B" | 6 |
group_by_dynamic#
group_by_dynamic(index) は、index 列の値に基づいて時間窓でグループ化します。時系列のダウンサンプリングで中心になる関数です。group_by() との根本的な違いは、グループのキーの出どころにあります。キーはデータにある列ではなく、時間軸を等間隔の升目に切ったもので、升目 1つが 1グループになります。最初のグループの開始点 start は、start_by、every、offset と index の先頭の値 value から決まります。start_by='window'（既定）では次の式になります。
start = value - (value - offset) % every
start_by='datapoint' なら、start は先頭の値そのものです。開始点が決まれば、各グループの範囲は次のように並びます。period の既定値は every と同じです。
[start, start + period)
[start + every, start + every + period)
...
[start + i * every, start + i * every + period)
グループの開始値がキーになります。include_boundaries=True を指定すると、_lower_boundary と _upper_boundary の 2列が追加され、各グループの開始（含む）と終了（含まない）が分かります。
次の例は、index の値を 3単位ごとにまとめて集計します。
index = [1, 2, 4, 5, 6, 8, 9, 10, 11]
# index は昇順に並べておく（set_sorted でソート済みと宣言）
df = pl.DataFrame(dict(index=index, value=list(range(len(index))))).set_sorted("index")
g = df.group_by_dynamic("index", every="3i", offset="0i", include_boundaries=True)
row(df, g.agg(pl.col("index").alias("index_in_group")))
shape: (9, 2)
|
shape: (4, 4)
|
結果を目で追いやすくするため、本書では plot_group_by_dynamic() でグループの区切りを可視化します。青い線が index の値、赤い線がグループのキー、緑の帯が各グループの範囲です。
from helper import utils
from helper.utils import plot_group_by_dynamic
from matplotlib import pyplot as plt
plt.ioff()
plot_group_by_dynamic(index, every='3i', offset='0i')
[[1, 2], [4, 5], [6, 8], [9, 10, 11]]
次に、offset 引数の影響を示すグラフを出力します。
plot_group_by_dynamic(index, every='3i', offset='1i')
[[1, 2], [4, 5, 6], [8, 9], [10, 11]]
plot_group_by_dynamic(index, every='3i', offset='2i')
[[1], [2, 4], [5, 6], [8, 9, 10], [11]]
plot_group_by_dynamic(index, every='3i', offset='-1i')
[[1], [2, 4], [5, 6], [8, 9, 10], [11]]
start_by='datapoint' にすると、最初のグループの開始点が index の先頭の値に固定されるため、offset は効きません。
plot_group_by_dynamic(index, every='3i', start_by='datapoint')
[[1, 2], [4, 5, 6], [8, 9], [10, 11]]
period の既定値は every と同じです。そのため、グループ同士は重なりません。period を every より大きくすると窓が重なり、同じ index が複数のグループに入ります。
plot_group_by_dynamic(index, every='3i', period='5i')
[[1, 2, 4], [4, 5, 6], [6, 8, 9, 10], [9, 10, 11]]
次のコードでは、index 列に基づいて 1単位ずつずらしながら、3単位ごとにグループ化します。
g = df.group_by_dynamic('index', every='1i', period='3i', include_boundaries=True)
row(df, g.agg(pl.col("index").alias('index_in_group')))
shape: (9, 2)
|
shape: (11, 4)
|
ここまでは整数列で説明しましたが、実際は時系列データに対して使うことが多いです。every、period、offset には次の単位が使えます。
1ns（ナノ秒）、1us（マイクロ秒）、1ms（ミリ秒）1s（秒）、1m（分）、1h（時間）1d（日）、1w（週）、1mo（月）、1q（四半期）、1y（年）
rolling#
rolling(index) は、index 列の値に基づいてローリングウィンドウを作ります。period と offset で窓の位置を調整でき、offset の既定値は -period です。
窓の範囲は () 側が含まれず、[] 側が含まれます。
(index[0] + offset, index[0] + offset + period]
...
(index[i] + offset, index[i] + offset + period]
次の例は、直前の 2点を含む 3点の窓で集計します。
g = df.rolling('index', period='3i')
g.agg(pl.col("value"))
| index | value |
|---|---|
| i64 | list[i64] |
| 1 | [0] |
| 2 | [0, 1] |
| 4 | [1, 2] |
| 5 | [2, 3] |
| 6 | [2, 3, 4] |
| 8 | [4, 5] |
| 9 | [5, 6] |
| 10 | [5, 6, 7] |
| 11 | [6, 7, 8] |
次のコードでは、index 列に基づき、各データ点とその後の 2点を含む 3点のローリングウィンドウを作り、その中で集計します。
g = df.rolling('index', period='3i', offset='-1i')
g.agg(pl.col("value"))
| index | value |
|---|---|
| i64 | list[i64] |
| 1 | [0, 1] |
| 2 | [1, 2] |
| 4 | [2, 3, 4] |
| 5 | [3, 4] |
| 6 | [4, 5] |
| 8 | [5, 6, 7] |
| 9 | [6, 7, 8] |
| 10 | [7, 8] |
| 11 | [8] |
行番号で固定点数の窓を作りたいときは、int_range() で行番号の列を作ってから rolling() に渡します。次のコードは、各行の前 2行を含む 3行の窓で A の最大値と B の最小値を計算します。
df = pl.DataFrame(
dict(
A=[5, 3, 8, 9, 10, 1, 2, 4],
B=[10, 20, 10, 2, 3, 5, 0, 7]
)
)
(
df
.rolling(
pl.int_range(0, pl.len()).alias('index'),
period='3i',
offset='-2i'
)
.agg(
pl.col.A,
pl.col.B,
pl.col.A.max().alias('A_max'),
pl.col.B.min().alias('B_min'),
)
)
| index | A | B | A_max | B_min |
|---|---|---|---|---|
| i64 | list[i64] | list[i64] | i64 | i64 |
| 0 | [5, 3] | [10, 20] | 5 | 10 |
| 1 | [5, 3, 8] | [10, 20, 10] | 8 | 10 |
| 2 | [3, 8, 9] | [20, 10, 2] | 9 | 2 |
| 3 | [8, 9, 10] | [10, 2, 3] | 10 | 2 |
| 4 | [9, 10, 1] | [2, 3, 5] | 10 | 2 |
| 5 | [10, 1, 2] | [3, 5, 0] | 10 | 0 |
| 6 | [1, 2, 4] | [5, 0, 7] | 4 | 0 |
| 7 | [2, 4] | [0, 7] | 4 | 0 |
Expr.rolling() も用意されています。引数は DataFrame.rolling() と同じですが、列単位で書けるぶん柔軟です。式に付けると、その式の結果に対して窓が作られます。
df.with_row_index().select(
pl.col('A').rolling('index', period='3i', offset='-2i'),
pl.col('A').max().rolling('index', period='3i', offset='-2i').alias('A_max')
)
| A | A_max |
|---|---|
| list[i64] | i64 |
| [5, 3] | 5 |
| [5, 3, 8] | 8 |
| [3, 8, 9] | 9 |
| [8, 9, 10] | 10 |
| [9, 10, 1] | 10 |
| [10, 1, 2] | 10 |
| [1, 2, 4] | 4 |
| [2, 4] | 4 |
.rolling() は直前の式全体に適用される点に注意してください。
pl.col('A').diff().max().rolling(...)（❶）— まずAが窓に分割され、そのあと各窓の中で.diff().max()が計算されます。差分を先に計算してから窓をかける（❷）—
dA = pl.col('A').diff()を別の列として作り、その列に.max().rolling(...)を適用します。
この 2つは結果が違います。どちらを意図しているかを意識して書き分けてください。
df1 = df.select('A').with_row_index().with_columns(
pl.col('A').rolling('index', period='3i'),
pl.col('A').diff().rolling('index', period='3i').alias('dA_rolling'),
pl.col('A').diff().max().rolling('index', period='3i').alias('dA_max_rolling') #❶
)
df2 = df.select('A').with_row_index().with_columns(
pl.col('A').diff().alias('dA') #❷
).with_columns(
pl.col('dA').rolling('index', period='3i').alias('dA_rolling'),
pl.col('dA').max().rolling('index', period='3i').alias('dA_max_rolling') #❷
)
row(df1, df2)
shape: (8, 4)
|
shape: (8, 5)
|
CookBook#
値の変化に応じたグループ化と集計#
次のコードでは、g 列の値が変わるたびにグループを切り、各グループの先頭の g と v の値を取り出します。rle_id() を使うと、連続する同じ値に同じ ID が振られ、値が変わるたびに ID が更新されます。この ID がそのままグループの番号になります。
df = pl.DataFrame(
{
"g": [1, 1, 2, 2, 2, 1, 1, 0],
"v": list(range(8))
}
)
df.group_by(
pl.col('g').rle_id().alias('group_id')
).agg(
pl.col('g').first(),
pl.col('v')
)
| group_id | g | v |
|---|---|---|
| u32 | i64 | list[i64] |
| 0 | 1 | [0, 1] |
| 1 | 2 | [2, 3, 4] |
| 2 | 1 | [5, 6] |
| 3 | 0 | [7] |
グループ毎に番号付け#
c1 と c2 の値の組み合わせに、0から始まる番号を振ります。新しい組み合わせが現れるたびに 1ずつ増やします。これは複合キーを整数 ID に写す処理です。
Polars 2.0 では agg_groups() が削除されたため、以下では 3通りの実装を示します。違いは番号の出どころです。グループ化してから展開する、1組み合わせ 1行の表を作って結合する、カテゴリカル型の内部番号を借りる、の 3つです。
df = pl.DataFrame(
{
"c1": ["b", "a", "a", "b", "c", "a", "b"],
"c2": [2, 1, 1, 1, 1, 1, 1],
}
)
group_by で計算#
group_by("c1", "c2", maintain_order=True)—c1とc2の組み合わせでグループを作ります。agg(pl.col("row_index"))— 各行に振った行番号を、グループごとにリストとして集めます。Polars 2.0 ではagg_groups()が削除されたため、with_row_index()で行番号の列を作ってから集計します。explode("row_index")— リストを行に展開します。sort("row_index")— 元の行順に戻します。drop("row_index")— 作業用の列を落とします。
(
df.with_row_index("row_index")
.group_by("c1", "c2", maintain_order=True)
.agg(pl.col("row_index"))
.explode("row_index")
.sort("row_index")
.drop("row_index")
)
| c1 | c2 |
|---|---|
| str | i64 |
| "b" | 2 |
| "a" | 1 |
| "a" | 1 |
| "b" | 1 |
| "c" | 1 |
| "a" | 1 |
| "b" | 1 |
join で計算#
join を使っても同じ結果になります。
df.select('c1', 'c2'):dfからc1列とc2列だけを取り出します。
unique(maintain_order=True):c1とc2の組み合わせから、ユニークなペアを取り出します。maintain_order=Trueを指定すると、元の行順が保たれます。
with_row_index():データフレームに行番号を追加します。
df.join(..., on=["c1", "c2"]):元の
dfに、番号を振ったユニークな組み合わせをjoinで戻します。on=["c1", "c2"]は、2列をキーに結合する指定です。
df.join(
df.select('c1', 'c2')
.unique(maintain_order=True)
.with_row_index(),
on=["c1", "c2"]
)
| c1 | c2 | index |
|---|---|---|
| str | i64 | u32 |
| "b" | 2 | 0 |
| "a" | 1 | 1 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |
| "c" | 1 | 3 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |
カテゴリカル型で計算#
c1 と c2 の値を 1つの文字列にまとめ、カテゴリカル型（Categorical）を経て整数に変換しても同じ結果になります。
pl.format("{} - {}", "c1", "c2"):pl.format()でc1とc2の値を 1つの文字列に組み立てます。{}はプレースホルダで、c1とc2の値が順に入ります。ここでは"c1の値 - c2の値"の形の文字列ができます。例えば、
c1が"a"、c2が1なら、"a - 1"になります。
cast(pl.Categorical):cast(pl.Categorical)で、その文字列列をCategorical型に変換します。Categorical型は、値の種類が限られた列を効率よく格納できる型です。
to_physical():to_physical()で、列を内部の格納形式に変換します。Categorical型なら、各カテゴリに対応する整数値が取り出せます。
alias("index"):最後に、できた列を
indexと名付けます。
df.with_columns(
pl.format("{} - {}", "c1", "c2")
.cast(pl.Categorical)
.to_physical()
.alias("index")
)
| c1 | c2 | index |
|---|---|---|
| str | i64 | u32 |
| "b" | 2 | 0 |
| "a" | 1 | 1 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |
| "c" | 1 | 3 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |
構造体とカテゴリカル型で計算#
pl.format() が使えないときは、構造体列をカテゴリカル型に変換する手もあります。
pl.struct("c1", "c2"):pl.struct("c1", "c2")は、c1とc2を 1つの構造体（struct）にまとめます。列の組み合わせを 1つの単位として扱えます。
rank("dense"):rank("dense")は、c1とc2のペアに「密なランク（dense ranking）」を振ります。同じ値には同じ順位を付け、順位が飛ばないようにします（例えば、1位が2人の場合、次は2位になります）。
cast(pl.String):cast(pl.String)は、ランク付けした値が整数のままだと直接カテゴリカル型に変換できないため、いったん文字列型（String）に変換します。
df.with_columns(
pl.struct("c1", "c2")
.rank("dense")
.cast(pl.String)
.cast(pl.Categorical)
.to_physical()
.alias("index")
)
| c1 | c2 | index |
|---|---|---|
| str | i64 | u32 |
| "b" | 2 | 0 |
| "a" | 1 | 1 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |
| "c" | 1 | 3 |
| "a" | 1 | 1 |
| "b" | 1 | 2 |