Aggregate with count distinct: pandas / polars / JSON

問題設定: 日別の合計利用時間とユニークユーザ数

Exercise 1

ユーザのアクティビティログから「日別の合計利用時間 (duration の総和)」と「日別のユニークユーザ数 (user_id のdistinct数)」を一度に集計したい．入力例は次のようなレコード形式とします．

import numpy as np
import pandas as pd

df = pd.DataFrame(
    {
        "date": ["2013-04-01", "2013-04-01", "2013-04-01", "2013-04-02", "2013-04-02"],
        "user_id": ["0001", "0001", "0002", "0002", "0002"],
        "duration": [30, 15, 20, 15, 30],
    }
)
df

	date	user_id	duration
0	2013-04-01	0001	30
1	2013-04-01	0001	15
2	2013-04-01	0002	20
3	2013-04-02	0002	15
4	2013-04-02	0002	30

▶ 期待する出力

date	duration	user_id
2013-04-01	65	2
2013-04-02	45	1

ポイント

sum のように単純な集約だけでなく，同じグループに対して別カラムでdistinct countをかける必要がある
pandas, polars, jq (= JSON→JSON 変換) のどれで書いても1pass で集計できる

Pattern 1: pandas

DataFrame.groupby(...).agg({...}) にカラム→集約関数の辞書を渡すのが最も素直． distinct count は pd.Series.nunique を直接渡すか，lambda x: x.nunique() を渡せばよい．

df.groupby("date").agg({"duration": "sum", "user_id": pd.Series.nunique})

	duration	user_id
date
2013-04-01	65	2
2013-04-02	45	1

lambda でも同じ結果が得られます．

df.groupby("date").agg({"duration": "sum", "user_id": lambda x: x.nunique()})

	duration	user_id
date
2013-04-01	65	2
2013-04-02	45	1

注意点 !

np.sum を渡しても動作するが，pandas 2.x 以降では文字列 "sum" の方が高速かつ FutureWarning を回避できる
出力カラム名を変えたい場合は，.agg(total=("duration", "sum"), n_users=("user_id", "nunique")) のような named aggregation が読みやすい

出力カラムを明示する書き方

(
    df.groupby("date")
      .agg(total_duration=("duration", "sum"),
           n_unique_users=("user_id", "nunique"))
      .reset_index()
)

	date	total_duration	n_unique_users
0	2013-04-01	65	2
1	2013-04-02	45	1

Pattern 2: polars

polars では group_by(...).agg([...]) の中に式 (expression)を並べる形になる． distinct count は pl.col(...).n_unique() で表現する．

import polars as pl

df_pl = pl.from_pandas(df)

df_pl.group_by("date").agg(
    pl.col("duration").sum().alias("total_duration"),
    pl.col("user_id").n_unique().alias("n_unique_users"),
).sort("date")

shape: (2, 3)

date	total_duration	n_unique_users
str	i64	u32
"2013-04-01"	65	2
"2013-04-02"	45	1

pandas との対応関係

操作	pandas	polars
group化	`groupby("date")`	`group_by("date")`
合計	`("duration", "sum")`	`pl.col("duration").sum()`
distinct count	`("user_id", "nunique")`	`pl.col("user_id").n_unique()`
出力カラム名のリネーム	named aggregation	`.alias(...)`

注意点 !

polars の group_by は結果の行順を保証しないため，期待する順序があれば .sort("date") を明示する
大規模データでは lazy() → .collect() を挟むことでクエリ最適化が効く

Pattern 3: JSON → JSON (jq)

ログを直接 JSON で扱いたい場面 (例: BigQuery export, Cloud Logging, fluentd) では jq 1本で同等の集計ができる． group_by → map で「グループごとに新しいオブジェクトを生成」というのが基本パターン．

入力 JSON

[
  {
    "date": "2013-04-01",
    "user_id": "0001",
    "duration": 30
  },
  {
    "date": "2013-04-01",
    "user_id": "0001",
    "duration": 15
  },
  {
    "date": "2013-04-01",
    "user_id": "0002",
    "duration": 20
  },
  {
    "date": "2013-04-02",
    "user_id": "0002",
    "duration": 15
  },
  {
    "date": "2013-04-02",
    "user_id": "0002",
    "duration": 30
  }
]

jq による集計

jq 'group_by(.date)
    | map({
        date: .[0].date,
        total_duration: (map(.duration) | add),
        n_unique_users: ([.[].user_id] | unique | length)
      })' activity.json

[
  {
    "date": "2013-04-01",
    "total_duration": 65,
    "n_unique_users": 2
  },
  {
    "date": "2013-04-02",
    "total_duration": 45,
    "n_unique_users": 1
  }
]

式の意味

group_by(.date) で date ごとの配列の配列に変換する (pandas の groupby 相当)
map({...}) でグループごとに新しいオブジェクトを組み立てる
map(.duration) | add で duration の合計を計算 (sum 相当)
[.[].user_id] | unique | length で user_id の distinct count を計算

注意点 !

unique はソート + 重複除去を行うため，要素数が多い場合はメモリを使う．巨大なログでは streaming 系 (--stream, gojq, duckdb の read_json + group by) に切り替えた方が安全
改行区切り JSON (NDJSON, JSONL) を扱う場合は，先頭で [inputs] + --slurp --null-input を組み合わせて配列化するのが定石

まとめ

3パターンの使い分け

pandas: 探索的分析・notebook での集計はこれが第一選択．named aggregation でカラム名まで一気に決められる
polars: 数百MB〜数GB級のログを高速に処理したいとき．式ベースなので集約ロジックの再利用がしやすい
jq: ファイル/ストリームで JSON が降ってくるシェル前段．DataFrame を経由せずに gh api / bq の出力を整形する場面で強い