Polarsの実行計画を読んでみよう。pandasとの性能差をGoogle Colabで検証する

以前、SQL / pandas / Polars 構文比較表 という記事で、SQL・pandas・Polarsの「書き方」の違いを比較しました。

今回はその続きとして、Polarsがなぜ速いのかを、Google Colab上で実際にコードを実行しながら検証します。

Pythonで表形式データを扱うライブラリとして、長く使われてきたのがpandasです。一方、近年は高速なデータ処理ライブラリとしてPolarsも注目されています。実際、データ量が増えるほど、pandasとPolarsの実行速度・パフォーマンスには大きな差がつく場合もあります。

理由としてよく挙げられるのは、次のような点です。

• Rustで実装されている
• 複数のCPUコアを利用できる
• 列指向のデータ形式(Arrow)を利用している
• Lazy APIによって処理を最適化できる

ただ、「Rustだから速い」「並列処理だから速い」というだけでは、半分くらいしか説明できていない気がしています。

PolarsのLazy APIでは、書いた処理をすぐには実行せず、いったん実行計画として保持します。そのあとで、不要な列を読み込まない・フィルタを早い段階でかける、といった最適化を行ってから、実際の処理を開始します。この実行計画は、explain()というメソッドで覗くことができます。

本記事では、Google Colab上に少し大きめのCSVを生成し、

1. pandas
2. PolarsのEager API
3. PolarsのLazy API

の3通りで同じ集計処理を書き、実行時間を比較します。あわせて、Polarsの実行計画を読みながら、速度差の理由を確認していきます。

セルは上から順に実行していけば、そのまま最後まで通る構成にしています。

対象読者

• pandasを使ったことがある
• Polarsに興味がある、あるいは触りはじめた
• Polarsが速いと聞くが、理由をふんわりとしか説明できない
• SQLの実行計画(EXPLAIN)に興味がある
• 大きめのCSVファイルを効率よく処理したい

Polarsの詳しい文法を知らなくても読み進められるよう、なるべくpandasと対比させながら進めます。細かい引数の仕様などは、Polarsのバージョンによって変わることがあるため、本記事では深入りせず、考え方の説明を中心にしています。

検証結果のはこちら

Google Colabでの検証過程は下記にあります。下記から、検証過程を再現することもできます。

今回実行する処理

題材として、次の2つのCSVを使います。

取引データ

列名	内容
transaction_id	取引ID
customer_id	顧客ID
transaction_date	取引日
amount	取引金額
status	取引状態
memo	今回の集計では使わない列

顧客マスタ

列名	内容
customer_id	顧客ID
customer_name	顧客名
segment	顧客区分
is_active	有効な顧客か

この2つに対して、次の処理を行います。

1. 2026年以降の取引を抽出する
2. completedの取引だけを残す
3. 取引金額が10,000円以上の行を残す
4. 有効な顧客だけを残す
5. 顧客IDで2つのデータを結合する
6. 顧客区分ごとに件数と取引金額を集計する
7. 取引金額の大きい順に並べる

フィルタ・Join・Group By・Sortをひと通り含む処理なので、実行計画を観察する題材としてもちょうどよいと思います。

Google Colabの準備

まずは必要なライブラリをインストールします。

!pip install -q polars pandas psutil

バージョンと実行環境も、記録として表示しておきます。ベンチマーク結果はCPUやメモリの状況で変わるため、あとで見返せるように残しておくと安心です。

import os
import platform

import pandas as pd
import polars as pl
import psutil

print("Python :", platform.python_version())
print("pandas :", pd.__version__)
print("Polars :", pl.__version__)
print("CPU    :", os.cpu_count())
print("RAM    :", round(psutil.virtual_memory().total / 1024**3, 2), "GB")

今回、Google Colabの無料枠で実行したところ、次のような環境になりました。

Python : 3.12.13
pandas : 2.2.2
Polars : 1.35.2
CPU    : 2
RAM    : 12.67 GB

CPUが2コアしかない、比較的控えめな環境です。有償ランタイムやローカルのマルチコア環境であれば、この後の結果はさらに変わってくるはずです。

大量のCSVファイルを生成する

Google Colabで試しやすいように、初期値は500万件にしています。手元の環境に応じてN_ROWSを変更してください。

N_ROWS = 5_000_000
N_CUSTOMERS = 200_000
CHUNK_SIZE = 500_000

TRANSACTIONS_PATH = "/content/transactions.csv"
CUSTOMERS_PATH = "/content/customers.csv"

# 少量で動作だけ確認したい場合
# N_ROWS = 1_000_000
# 通常のベンチマーク
# N_ROWS = 10_000_000
# 高RAM環境向け(後述)
# N_ROWS = 100_000_000

1億件規模のDataFrameを一度にメモリへ作ろうとすると、CSVを書き出す前にメモリ不足になることがあります。そこで、一定件数ずつ生成してCSVへ追記していく形にしています。

まずは顧客マスタです。

import numpy as np

rng = np.random.default_rng(42)

customers = pd.DataFrame(
    {
        "customer_id": np.arange(1, N_CUSTOMERS + 1, dtype=np.int32),
        "customer_name": [f"customer_{i:06d}" for i in range(1, N_CUSTOMERS + 1)],
        "segment": rng.choice(
            ["individual", "small", "enterprise"],
            size=N_CUSTOMERS,
            p=[0.7, 0.2, 0.1],
        ),
        "is_active": rng.choice([True, False], size=N_CUSTOMERS, p=[0.9, 0.1]),
    }
)
customers.to_csv(CUSTOMERS_PATH, index=False)
customers.head()

続いて取引データです。CHUNK_SIZE件ずつ生成し、追記モードでCSVへ書き出します。

if os.path.exists(TRANSACTIONS_PATH):
    os.remove(TRANSACTIONS_PATH)

base_date = np.datetime64("2024-01-01")

for start in range(0, N_ROWS, CHUNK_SIZE):
    size = min(CHUNK_SIZE, N_ROWS - start)
    transaction_dates = base_date + rng.integers(0, 365 * 3, size=size).astype("timedelta64[D]")

    chunk = pd.DataFrame(
        {
            "transaction_id": np.arange(start + 1, start + size + 1, dtype=np.int64),
            "customer_id": rng.integers(1, N_CUSTOMERS + 1, size=size, dtype=np.int32),
            "transaction_date": transaction_dates,
            "amount": rng.integers(100, 100_000, size=size, dtype=np.int32),
            "status": rng.choice(["completed", "canceled"], size=size, p=[0.85, 0.15]),
            "memo": "unused transaction memo",
        }
    )
    chunk.to_csv(TRANSACTIONS_PATH, mode="a", header=(start == 0), index=False)
    print(f"{start + size:,} / {N_ROWS:,} rows")

transactions_size = os.path.getsize(TRANSACTIONS_PATH) / 1024**3
customers_size = os.path.getsize(CUSTOMERS_PATH) / 1024**2

print(f"transactions.csv: {transactions_size:.2f} GB")
print(f"customers.csv   : {customers_size:.2f} MB")

実際に生成すると、次のサイズになりました。

transactions.csv: 0.30 GB
customers.csv   : 7.16 MB

500万件でも0.3GB程度になるので、memoのような使わない列を読み飛ばせるかどうかが、地味に効いてきそうな規模感です。なお、CSVの生成時間はこのあとのベンチマークには含めません。あくまで「同じ入力データを使う」ための下準備です。

実行時間を測定する関数

3つの処理を同じ形式で計測できるよう、簡単なヘルパーを用意します。

import gc
import time


def measure(label, function):
    gc.collect()
    start = time.perf_counter()
    result = function()
    elapsed = time.perf_counter() - start
    print(f"{label}: {elapsed:.2f}秒")
    return result, elapsed

pandasで処理する

まずはpandasで、今回の処理を素直に書いてみます。

def run_pandas():
    transactions = pd.read_csv(TRANSACTIONS_PATH, parse_dates=["transaction_date"])
    customers = pd.read_csv(CUSTOMERS_PATH)

    filtered_transactions = transactions[
        (transactions["transaction_date"] >= "2026-01-01")
        & (transactions["status"] == "completed")
        & (transactions["amount"] >= 10_000)
    ]
    active_customers = customers[customers["is_active"]]

    joined = filtered_transactions.merge(active_customers, on="customer_id", how="inner")

    result = (
        joined.groupby("segment", as_index=False)
        .agg(
            transaction_count=("transaction_id", "count"),
            total_amount=("amount", "sum"),
        )
        .sort_values("total_amount", ascending=False)
    )
    return result

pandas_result, pandas_time = measure("pandas", run_pandas)
pandas_result

500万件のデータに対して実行すると、次のようになりました。

pandas: 8.08秒

      segment  transaction_count  total_amount
1  individual             802417   44120902892
2       small             227616   12521731951
0  enterprise             115051    6314748701

このコードでは、取引CSV全体をいったんDataFrameとして読み込んでから、フィルタやJoinをかけています。pandasにもusecolsやchunksizeなど大規模CSV向けの機能はありますが、今回はあえて使わず、「素直に書いた場合」の挙動を見ています。

PolarsのEager APIで処理する

次に、PolarsのEager API(呼び出すたびにその場で評価される書き方)で同じ処理を書きます。

def run_polars_eager():
    transactions = pl.read_csv(
        TRANSACTIONS_PATH,
        schema_overrides={"transaction_date": pl.Date},
    )
    customers = pl.read_csv(CUSTOMERS_PATH)

    filtered_transactions = transactions.filter(
        (pl.col("transaction_date") >= pl.date(2026, 1, 1))
        & (pl.col("status") == "completed")
        & (pl.col("amount") >= 10_000)
    )
    active_customers = customers.filter(pl.col("is_active"))

    result = (
        filtered_transactions.join(active_customers, on="customer_id", how="inner")
        .group_by("segment")
        .agg(
            pl.len().alias("transaction_count"),
            pl.col("amount").sum().alias("total_amount"),
        )
        .sort("total_amount", descending=True)
    )
    return result

polars_eager_result, polars_eager_time = measure("Polars Eager", run_polars_eager)
polars_eager_result

同じ条件で実行すると、3.75秒でした。pandasの8.08秒に対して、Eager APIに書き換えるだけでもおよそ2.2倍速くなっています。集計結果自体はpandasとぴったり一致しています。

Polars Eager: 3.75秒

Eager APIの時点でも、pandasより速くなることがあります。ただしこのコードでは、read_csv()でCSV全体を読み込んでから処理しているため、Polars本来の最適化はまだ活きていません。

PolarsのLazy APIで処理する

Lazy APIでは、read_csv()の代わりにscan_csv()を使います。この時点ではCSVはまだ読み込まれず、「どこから読むか」という情報だけが保持されます。

そのあとに書くフィルタ・Join・集計も、その場では実行されず、ひとまとまりの実行計画として積み上がっていきます。実際に処理が走るのは、最後にcollect()を呼んだタイミングです。

transactions_lazy = pl.scan_csv(
    TRANSACTIONS_PATH,
    schema_overrides={"transaction_date": pl.Date},
)
customers_lazy = pl.scan_csv(CUSTOMERS_PATH)

lazy_query = (
    transactions_lazy.filter(
        (pl.col("transaction_date") >= pl.date(2026, 1, 1))
        & (pl.col("status") == "completed")
        & (pl.col("amount") >= 10_000)
    )
    .join(
        customers_lazy.filter(pl.col("is_active")),
        on="customer_id",
        how="inner",
    )
    .group_by("segment")
    .agg(
        pl.len().alias("transaction_count"),
        pl.col("amount").sum().alias("total_amount"),
    )
    .sort("total_amount", descending=True)
)

この時点では、まだ何も計算されていません。lazy_queryには「これから何をするか」という手順だけが入っています。実際に計算するにはcollect()を呼び出します。

def run_polars_lazy():
    return lazy_query.collect()


polars_lazy_result, polars_lazy_time = measure("Polars Lazy", run_polars_lazy)
polars_lazy_result

結果は1.55秒でした。同じ処理・同じデータにもかかわらず、scan_csv() + Lazy APIに変えるだけで、Eager APIよりさらに半分以下まで縮んでいます。

Polars Lazy: 1.55秒

実行時間を比較する

3つの実行時間をまとめて見てみます。

benchmark = pd.DataFrame(
    {
        "library": ["pandas", "Polars Eager", "Polars Lazy"],
        "seconds": [pandas_time, polars_eager_time, polars_lazy_time],
    }
)
benchmark["relative_to_pandas"] = benchmark["seconds"] / pandas_time
benchmark

実行結果は次のとおりです。

処理方法	実行時間	pandasとの比較
pandas	8.08秒	1.00
Polars Eager	3.75秒	0.46
Polars Lazy	1.55秒	0.19

CPU2コアというかなり控えめな環境でも、Polars LazyはpandasのおよそD5倍、Eager APIと比べても2倍以上速いという結果になりました。もちろんこの数値は環境やデータ内容に左右されるので、絶対値そのものよりも「Eager→Lazyで further speedupが出ている」という傾向を見るのが本題です。

集計結果が一致しているか確認する

速度を比べる前提として、そもそも同じ結果になっているかを確認しておきます。

pandas_check = pandas_result.sort_values("segment").reset_index(drop=True)
polars_check = polars_lazy_result.sort("segment").to_pandas().reset_index(drop=True)

pd.testing.assert_frame_equal(pandas_check, polars_check, check_dtype=False)
print("集計結果は一致しています")

実行すると、次の通り一致が確認できました。

集計結果は一致しています

Polarsの実行計画を表示する

ここからが本題です。Polarsではexplain()を使って、LazyFrameの実行計画を文字列として確認できます。optimized引数で、最適化前・最適化後を切り替えられます。

print(lazy_query.explain(optimized=False))

実際に出力すると、次のようになりました。

SORT BY [descending: [true]] [col("total_amount")]
  AGGREGATE[maintain_order: false]
    [len().alias("transaction_count"), col("amount").sum().alias("total_amount")] BY [col("segment")]
    FROM
    INNER JOIN:
    LEFT PLAN ON: [col("customer_id")]
      FILTER [(col("amount")) >= (10000)]
      FROM
        FILTER [(col("status")) == ("completed")]
        FROM
          FILTER [(col("transaction_date")) >= (2026-01-01)]
          FROM
            Csv SCAN [/content/transactions.csv]
            PROJECT */6 COLUMNS
            ESTIMATED ROWS: 5356547
    RIGHT PLAN ON: [col("customer_id")]
      FILTER col("is_active")
      FROM
        Csv SCAN [/content/customers.csv]
        PROJECT */4 COLUMNS
        ESTIMATED ROWS: 218874
    END INNER JOIN

コードに書いた通り、CSVスキャンのあとにフィルタが3段(FILTER)、そのあとにJoin・集計・Sortという順で並んでいます。列もPROJECT */6 COLUMNSとなっていて、この時点では全列を読む前提になっています。

print(lazy_query.explain(optimized=True))

最適化後は、次のように変わりました。

SORT BY [descending: [true]] [col("total_amount")]
  AGGREGATE[maintain_order: false]
    [len().alias("transaction_count"), col("amount").sum().alias("total_amount")] BY [col("segment")]
    FROM
    simple π 2/2 ["amount", "segment"]
      INNER JOIN:
      LEFT PLAN ON: [col("customer_id")]
        Csv SCAN [/content/transactions.csv]
        PROJECT 4/6 COLUMNS
        SELECTION: [([([(col("amount")) >= (10000)]) & ([(col("transaction_date")) >= (2026-01-01)])]) & ([(col("status")) == ("completed")])]
        ESTIMATED ROWS: 5356547
      RIGHT PLAN ON: [col("customer_id")]
        Csv SCAN [/content/customers.csv]
        PROJECT 3/4 COLUMNS
        SELECTION: col("is_active")
        ESTIMATED ROWS: 218874
      END INNER JOIN

注目したいのは次の2点です。

• 取引データのPROJECT */6 COLUMNSがPROJECT 4/6 COLUMNSに変わり、使わないmemo列などが読み込み対象から外れている
• 最適化前は独立していた3つのFILTERが、SELECTIONとしてCSVスキャンの直下にまとめられている

実行計画は、CSVを読み込む部分から最終結果に向かって、下から上に追っていくとイメージしやすくなります。

transactions.csv / customers.csv をスキャン
        ↓
それぞれにフィルタを適用
        ↓
customer_idでJoin
        ↓
segment単位で集計
        ↓
total_amountで並べ替え

Projection Pushdownを読む

最適化後の実行計画では、取引データ側がPROJECT 4/6 COLUMNSになっていました。今回の取引データにはtransaction_id / customer_id / transaction_date / amount / status / memoの6列がありますが、memoはフィルタ・Join・集計のどこにも使われていません。

そのためPolarsは、「memo列はそもそも読み込まなくてよい」と判断できます。この最適化をProjection Pushdownと呼びます。

不要な列をいったんDataFrameへ読み込んでから捨てるのではなく、できる限りスキャンの時点で読み込み対象から外す、という考え方です。とくにmemoのようなサイズの大きい文字列列が含まれる場合、この差はデータ量が増えるほど効いてきます。

Predicate Pushdownを読む

次に注目したいのがフィルタの位置です。最適化前の実行計画では、FILTERが3段に分かれてCSVスキャンより上に置かれていましたが、最適化後はSELECTIONとしてCSVスキャンの直下、つまりスキャンとほぼ同時に評価される位置まで引き上げられていました。この最適化をPredicate Pushdownと呼びます。

たとえば取引データが1億件あり、条件に合う行が300万件しかない場合を考えます。フィルタをJoinのあとにかけると、Joinは1億件同士の突き合わせになりますが、フィルタを先にかけられれば、Joinに渡るのは300万件だけになります。Joinへ渡す件数が減れば、それだけJoin自体の負荷も下がります。

今回の500万件のデータでも、ESTIMATED ROWS: 5356547(取引側)・ESTIMATED ROWS: 218874(顧客側)という見積もりが、フィルタ適用前の全件に近い数値になっている点は少し気になりますが、実際にJoinへ渡る件数はフィルタ後まで絞り込まれた状態になっています。

Joinの前に両方のデータを小さくする

今回は取引データだけでなく、顧客マスタ側にもis_activeのフィルタを書いています。そのため、Polarsは次のような計画を立てられます。

transactions.csv --フィルタ--> 必要な取引だけ ─┐
                                              ├─ Join
customers.csv    --フィルタ--> 有効な顧客だけ ─┘

Joinが速いかどうかは、Join処理そのものの実装だけでなく、「Joinの手前で両側のデータをどれだけ小さくできているか」にも左右されます。Polarsの実行計画を読むときは、Joinノードだけでなく、その直前で何件のデータが渡っているかにも注目すると理解が深まります。

profileで実際の処理時間を見る

explain()はあくまで「立てた計画」を見るものです。実際にどのノードにどれくらい時間がかかったかは、profile()で確認できます。

profile_result, profile = lazy_query.profile()
profile

実行すると、次のような内訳が得られました(単位はマイクロ秒)。

shape: (5, 3)
┌─────────────────────────────────┬─────────┬─────────┐
│ node                            ┆ start   ┆ end     │
│ ---                             ┆ ---     ┆ ---     │
│ str                             ┆ u64     ┆ u64     │
╞═════════════════════════════════╪═════════╪═════════╡
│ optimization                    ┆ 0       ┆ 2228981 │
│ join(customer_id)               ┆ 2228981 ┆ 2650115 │
│ simple-projection(amount, segm… ┆ 2650426 ┆ 2650447 │
│ .streaming_group_by()           ┆ 2650991 ┆ 2711895 │
│ sort(total_amount)              ┆ 2711906 ┆ 2712005 │
└─────────────────────────────────┴─────────┴─────────┘

見ると、最初の区間(CSVスキャン+フィルタ込みと思われる部分)だけでおよそ2.23秒かかっていて、Join(約0.42秒)・Group By(約0.06秒)・Sort(0.0001秒程度)は相対的に軽い処理であることが分かります。なお、profile()はプロファイリング自体のオーバーヘッドがあるため、合計はcollect()単体で計測した1.55秒よりやや長め(約2.71秒)になっています。それでも、CSVスキャン周りが最も時間を要する処理であるという傾向は、この結果からも読み取れます。

環境によっては、グラフとして表示することもできます。

profile_result, profile = lazy_query.profile(show_plot=True)

profile()の出力では、おおまかに次のような点を確認します。

• CSVスキャンに時間がかかっているか
• Joinに時間がかかっているか
• Group Byに時間がかかっているか
• Sortに時間がかかっているか

explain()で処理の「構造」を、profile()で「実測時間」を見る、という組み合わせで考えると、どこがボトルネックなのかを把握しやすくなります。

ストリーミングエンジンで実行する

データがメモリに収まりきらないほど大きい場合は、ストリーミングエンジンも選択肢になります。データ全体を一度にメモリへ載せず、バッチに分けて処理する方式です。

def run_polars_streaming():
    return lazy_query.collect(engine="streaming")


polars_streaming_result, streaming_time = measure("Polars Streaming", run_polars_streaming)

今回は1.46秒となり、通常のLazy実行(1.55秒)とほぼ同水準、わずかに速いという結果でした。

Polars Streaming: 1.46秒

500万件・0.3GB程度のデータであれば、そもそもメモリに十分収まる規模なので、ストリーミングにしたからといって劇的に速くなるわけではない、というのが今回の実測結果です。ストリーミングにすれば必ず速くなる、というわけではありません。データ量や処理内容によっては、通常のインメモリエンジンの方が速いこともあります。速度だけでなく、「メモリに収まらないデータを扱うための選択肢」として捉えておくとよさそうです。

なぜPolarsは速いのか

ここまでの実行計画と実測結果から、Polarsの速さは1つの理由だけでは説明できないことが分かります。今回の検証では、pandasの8.08秒に対し、Polars Eagerが3.75秒、Polars Lazyが1.55秒でした。同じ処理・同じデータでこれだけ差が出た背景には、次のような要素が組み合わさっています。

不要な列を読み込まない
Projection Pushdownにより、実際にPROJECT */6 COLUMNSがPROJECT 4/6 COLUMNSに変わり、集計に使わないmemo列などを、できるだけ早い段階で処理対象から外せていました。

フィルタを早く適用する
Predicate Pushdownにより、最適化前は独立していたFILTERが、CSVスキャン直下のSELECTIONにまとめられていました。全件をそのままJoinや集計に渡すのではなく、条件に合う行だけを後続処理に渡す形になっています。

小さくしてからJoinする
取引データ・顧客マスタの両方にフィルタをかけ、小さくなったデータ同士でJoinします。profile()を見ても、Join自体にかかった時間は0.42秒程度と、全体の一部にとどまっていました。

処理全体を見てから順序を決める
Lazy APIでは個々のメソッドをすぐには実行せず、一連の処理をまとめて最適化してから実行します。この「まとめて最適化してから実行する」という差が、EagerからLazyへの2倍以上の速度差として表れたと考えられます。

並列処理や列指向処理を活かす
実行計画によって処理対象を絞り込んだうえで、Polarsの実行エンジンが計算を行います。今回はCPU2コアという控えめな環境でもこの結果だったので、コア数が多い環境ではさらに差が広がる可能性があります。

つまりPolarsの速さは、「Rustで実装されているから」だけでなく、「そもそも不要なデータを処理していない」ことにも支えられている、と言えそうです。

pandasが使えないという話ではない

この結果だけを見て、「pandasは遅いから使うべきではない」と結論づけるのは早計だと思います。pandasには、

• 利用者が多く、情報やサンプルコードが豊富
• 対応しているライブラリが多い
• 小規模なデータでは十分に高速
• 対話的なデータ分析に使いやすい

といった長所があります。また、pandasでもusecolsで列を絞ったり、chunksizeで分割して読み込んだりすることは可能です。

違いは、pandasでは開発者が明示的に工夫する場面が多いのに対し、PolarsのLazy APIでは、処理全体を実行計画として保持し、オプティマイザが自動的に改善してくれる、という点にあると思います。今回はコードをほぼ書き換えず、read_csvをscan_csvに変えてcollect()するだけで、8.08秒→1.55秒という差が出た点は、この「自動的に改善してくれる」効果が大きいことを表していると思います。

SQLの実行計画との共通点

SQLも、書いた通りの順番でそのまま処理されているわけではなく、実行計画を作ってからインデックスの利用やJoinの順序を決めます。PolarsのLazy APIも、考え方としてはこれに近いものです。

result = (
    pl.scan_csv("transactions.csv")
    .filter(...)
    .join(...)
    .group_by(...)
    .agg(...)
    .collect()
)

これはPythonのメソッドチェーンとして書かれていますが、内部では1つのクエリとしてまとめて最適化されています。以前書いたSQL / pandas / Polars 構文比較表では「書き方」の違いを比較しましたが、今回の内容を踏まえると、Polarsは単なる高速なDataFrameライブラリというより、Pythonから使えるクエリエンジンに近い、と捉えると理解しやすいかもしれません。

より大きい件数で試す場合

N_ROWSを変更すれば、同じNotebookでより大きい件数を試せます。

N_ROWS = 100_000_000

ただし、Google Colabの通常ランタイムでは、CSVの生成やpandasでの一括読み込みの段階でメモリ不足になる可能性があります。今回は500万件(0.3GB程度)で完走を確認しましたが、まずは数百万件程度で処理全体を確認し、そのうえで環境に合わせて件数を増やしていくのがおすすめです。

pandasがメモリ不足で完走しなかった場合も、それ自体が1つの結果です。次のように、完走できたかどうかも含めて記録しておくと、あとから見返しやすくなります。

処理方法	500万件	1,000万件	1億件
pandas	完走(8.08秒)	実行後に記入	完走 or メモリ不足
Polars Eager	完走(3.75秒)	実行後に記入	実行後に記入
Polars Lazy	完走(1.55秒)	実行後に記入	実行後に記入
Polars Streaming	完走(1.46秒)	実行後に記入	実行後に記入

最速の実行時間だけでなく、件数を増やしたときにどこまで完走できるかも、大規模データ処理では見ておきたいポイントだと思います。

まとめ

今回は、Google Colab上に500万件・0.3GB程度のCSVを生成し、pandasとPolars(Eager / Lazy)で同じ集計処理を実行して比較しました。結果は、pandas 8.08秒、Polars Eager 3.75秒、Polars Lazy 1.55秒、Polars Streaming 1.46秒で、CPU2コアという控えめな環境でもLazy APIはpandasのおよそ5倍という速度になりました。あわせて、explain()でPolarsの実行計画を、profile()で実測時間を確認しました。

PolarsのLazy APIでは、主に次のような最適化が行われます。

• 不要な列を読み込まない(Projection Pushdown。今回はPROJECT */6 COLUMNS→PROJECT 4/6 COLUMNSという変化として実際に確認できました)
• フィルタを早い段階で適用する(Predicate Pushdown。3段のFILTERが、スキャン直下のSELECTIONにまとまりました)
• Joinへ渡すデータを小さくしてから結合する
• 処理全体を見てから実行順序を決める
• 必要に応じてストリーミング実行に切り替えられる

Polarsが速いのは、単に低水準言語で実装されているからではなく、実行前に処理全体を見渡し、「できるだけ少ないデータだけを処理する計画」に変換していることも、大きな要因だと思います。

「Polarsで実行したら速かった」で終わらせず、「不要な列が読み込まれず、フィルタがJoinより前に適用されたから速くなった」というところまで、実行計画を読みながら説明できるようになると、Polarsに限らず他のクエリエンジンを触るときにも応用が効くと思います。

さいごに

前回のSQL / pandas / Polars 構文比較表が「書き方」の比較だったのに対し、今回は「なぜ速いか」という一段掘り下げた内容になりました。

Google Colab上で、CSV生成からベンチマーク、実行計画の確認までを1つのNotebookに収められるのは、この手の検証を試すうえで地味に便利だと感じています。過去に書いたPython→Goのリファクタリングによる速度比較でも触れましたが、こうした比較検証は、単純な速度差だけでなく、その裏にある設計・最適化の考え方まで見ていくと、得られるものが多いように思います。

弊社について

本記事を書いている 合同会社インクルーシブソリューションズ は、データ基盤構築・分析基盤設計・システム改善支援を中心に活動している小規模IT法人です。

主な領域は、

• データマート設計・データパイプライン構築
• SQL / Python を用いたデータ処理設計
• BI導入支援・分析基盤の整備
• 既存システムの運用改善・可視化支援

といった、「データを使える状態にする」ための活動です。

弊社の企業活動に興味がある方は、ぜひ公式サイトも覗いてみてください。