Pandasは絶滅すべきである

Pandasは絶滅すべきだ

その通り、Pandasは絶滅すべきです。国際外交に使われる愛らしいフワフワした生き物ではなく、PythonのDataFrameライブラリのことです。

なぜか？ Pandasの非効率性が、ワークロードがその複雑さを正当化する前に、分散クエリシステムを採用することを強制するからです。ほとんどのワークロードはそれらのシステムを正当化せず、それらは単によくマーケティングされた「銀の弾丸」に過ぎないと私は主張します。

私が何を言っているのかを理解するために、まずPandasの典型的な採用経路を理解する必要があります。

なぜPandasを使うのか？

下の図は、作業しているデータサイズに基づいて、各DataFrameライブラリを採用することを検討する一般的な時期の目安を示しています。左から右へたどると、データ分析ツールの典型的な採用経路と、Pandasユーザーがあるデータサイズを超えたときに経験する崖もわかります。

人々は通常Excelから始め、GBの範囲でPandasに移行します。PandasはGB台の数十までうまく機能しますが、その後メモリの問題、計算速度の低下、またはPandasの難解なAPIに不満を感じ始めます。この時点での従来の答えは、Spark、DataBricks、Snowflake、またはDaskのような、ビッグデータ™️向けに設計された「本物の」（つまり高価な）ツールに移行することです。

［画像: データサイズ別の一般的なDataFrameライブラリの採用比較］

データサイズ別のDataFrameライブラリ採用の目安

ここで重要なのは、「Pandasの崖」と、分散システムが本当に必要となるスケールの間には、成長中のギャップがあるということです。このギャップは、約100GBのマークにあり、現代的で高性能な単一マシンツールで効果的に埋めることができます。私は主にPolarsとDuckDBについて話しています。

なぜこの約100GBのしきい値がそれほど重要なのでしょうか？ その答えは、世の中にどれだけの「ビッグデータ」が存在するかを理解することにあります。

私にはビッグデータがある、そうだろう？

2024年、Amazonは「TPCは十分ではない：Amazon Redshiftフリートの分析」と題する論文を発表しました。この論文の目的は、Amazon自身の分散分析データベースであるRedshiftからのテレメトリデータと、業界標準のデータベースベンチマークで使用されるクエリパターンを比較することでした。分析の一環として、Amazonはフリートのクエリ実行時間とテーブルサイズの統計情報を公開しました。

［画像: Amazon Redshiftフリートのクエリ実行時間のバケット別集計］

Amazon Redshiftフリートのクエリ実行時間のバケット別集計

［画像: Amazon Redshiftフリートのテーブルサイズのバケット別集計］

Amazon Redshiftフリートのテーブルサイズのバケット別集計

いくつかの仮定を置けば、Amazonの顧客が分析データベースをどのように使用しているかについて、興味深い結論を導き出すことができます。以下のように仮定しましょう：

• Redshiftテーブルの行の平均サイズは1KBである
• 各RedShiftクラスターは10台のマシンで構成されており、各マシンはS3から8GB/sのデータを取り込むことができ、それ以外のことはしない

すると、以下のことがわかります：

• Redshiftフリートのテーブルの94.68%は100GB未満のデータを含んでいます
• クエリの86.9%は80GB以下のデータで動作しています

このデータセットについて、さらに深く掘り下げたい場合は、MotherDuckのJordan Tiganiがこちらで詳細な分析を行っています。注意：MotherDuckはDuckDBホスティングを販売するSaaSビジネスであるため、ある程度の懐疑心は必要かもしれません。

計算の説明

実行時間テーブルの最初の3行を合計すると、クエリの86.9%が1秒未満で実行されると計算できます。

10台のマシンが8GB/sの速度でデータを処理するという仮定を使用して、次のように計算します：

    10 machines * 8GB * 1 second = 80GB of data

8GB/sの仮定は、この認めるところ古くなったベンチマークに基づいています。

「94.68%のテーブルが100GB未満を含む」という主張に到達するには、10^8までの行を合計し、94.68%の行を得ます。次に、行あたり1KBという仮定を使用して計算します：

    10^8 rows in a table * 1KB = 100GB

おそらく、1KB/行という仮定は楽観的すぎますが、たとえ10KBを仮定しても、テーブルサイズは1TBに達します。

しかし、これはすべて何を意味するのか？

あなたはおそらくビッグデータを持っていないし、おそらく将来も持たないでしょう。あなたはミディアムデータの問題を抱えており、ミディアムデータソリューションを必要としています。

代替案を紹介します

前述したように、私が提案する代替案はDuckDBとPolarsです。大まかに言うと、PolarsはRustベースのDataFrameライブラリで、Pandasに似ていますが、いくつかの重要な点で異なります。DuckDBはインメモリ分析DBであり、本質的には分析用のSQLiteです。これらのツールとPandasとの違いを理解するために、例を見てみましょう。

10億行チャレンジは、10億行の天気ステーションデータを含むCSVの最小値、平均値、最大値を計算する最速のJavaプログラムを書くためのチャレンジでした。コンペティションで受理された最速の実装は1.5秒で実行されました。

元のチャレンジでは、ベアメタルのHetzner AX161サーバー（32コア、128GB RAM、Debian 12実行）が使用されました。著者はシリアルな先延ばし癖があり、ケチで、Hetznerでは大きなサーバーを借りるために「評判」を築く必要があるため、代わりにAWSのm7a.8xlargeがこれらのテストに使用されました。これもDebian 12を実行しています。

この基本的な構成は、元のチャレンジと同じです：32コア、128GB RAM、AMD CPU。ただし、ベアメタル専用ハードウェアを使用しないことは、再現性にいくらか影響を与える可能性があります（すみません）。

黙ってコードを見せてくれ

早速ですが、いくつかの実装を見てみましょう。

Pandas

Pandasに触れたことがある人なら誰でも見慣れたはずです。CSVからデータを読み込み、天気ステーションでグループ化し、最小値、平均値、最大値の集計を計算します。

出力のシリアライゼーションはどこ？

パフォーマンステストでは、元のチャレンジで指定された出力シリアライゼーションはスキップされます。実装にはすべて出力のシリアライズ機能が含まれており、これは実装の単体テストに使用されました（例：Pandasコード）。10億行チャレンジの出力形式は標準的ではないため、シリアライゼーションのテストはさまざまなライブラリの関連テストとは見なされませんでした。

def do_1brc_pandas(file_path: str):
    df = (
        pd.read_csv(file_path, sep=";", names=["station", "measurement"])
        .groupby("station")
        .agg({"measurement": ["min", "mean", "max"]})
        .round(2)
    )

この例で覚えておくべき重要な点は、Pandasがこの計算の各ステップを逐次的かつ即座に実行することです。データセット全体を読み込み、グループ化してから集計を実行します。

Polars

PolarsのコードはPandasに似ていますが、実行時には大きく異なります。後述します。

def do_1brc_polars(file_path: str):
    df = (
        pl.scan_csv(
            file_path,
            separator=";",
            new_columns=["station", "measurement"],
            has_header=False,
        )
        .group_by("station")
        .agg(
            pl.col("measurement").min().round(2).alias("min"),
            pl.col("measurement").mean().round(2).alias("mean"),
            pl.col("measurement").max().round(2).alias("max"),
        )
        .collect(new_streaming=True) # Stream the input data and perform computations in chunks
    )

データはチャンクでスキャンされ、グループ化され、集計されます。ここで重要なのは、scan_csvは遅延評価され、.collectの呼び出しがクエリパイプラインを実行することです。これはデータベース用語のように聞こえるかもしれませんが、そうであるべきです。この遅延評価により、Polarsはデータベースと同様に最適化されたクエリグラフを構築し、40年分のデータベース最適化を活用して、データをチャンク単位で読み込み、必要に応じてスレッド間で作業を並列化できます。

データベースと同様に、.collectの呼び出しを.explain(streaming=True)と.explain(streaming=True, optimized=False)に置き換えることで、最適化されたクエリプランと最適化されていないクエリプランを視覚化できます。

最適化されたクエリプラン

このクエリプランはそれほどエキサイティングではありません。CSVをスキャンし、2列の射影を行い、集計します。フィルタリングを含むクエリでは、述語プッシュダウンが適用されることが期待されます。これは、行が集計前にフィルタリングされることを意味します。Pandasとは異なり、すべての行がメモリにロードされてからフィルタリングされます。

AGGREGATE
        [col("measurement").min().round().alias("min"), col("measurement").mean().round().alias("mean"), col("measurement").max().round().alias("max")] BY [col("station")] FROM
    STREAMING:
    simple π 2/2 ["measurement", "station"]
        Csv SCAN [/Users/eddie/Documents/code/pandas-should-go-extinct/data/measurements.csv]
        PROJECT 2/2 COLUMNS

DuckDB

DuckDBのコードは、通常のSQLのように記述されます。列は集計関数を適用して選択され、グループ化基準が指定されます。

def do_1brc_duckdb(file_path: str):
    df = duckdb.read_csv(file_path, names=["station", "measurements"])

    src = duckdb.sql("""
          create table src as 
          select 
            station, 
            min(measurements) min, 
            max(measurements) max, 
            cast(avg(measurements) as decimal(8, 1)) avg 
          from df 
          group by station
        """
    )

このコードサンプルの重要な点は、DuckDBが、データがどこにどのように格納されていても、データに対するSQLインターフェースを提供することです。また、メモリ内のPythonオブジェクトをクエリする機能も備えています。上記のリストでは、dfオブジェクトはCSVを読み込むことによって作成され、SQLを使用してクエリされます。

DuckDBはPolarsと同様に、クエリプランを構築して実行します。これは、クエリエンジンが適切と判断した場合、遅延的、マルチスレッド、チャンク単位で適用されます。.explain()を付け加えることで、DuckDBがクエリのために生成するクエリプランを表示することもできます。

最適化されたクエリプラン

ここでも、クエリプランはそれほどエキサイティングではありません。CSVをスキャンし、2列の射影を行い、集計します。

注意：これは私のM1 MBAでプランを実行したときの出力であり、以下のパフォーマンスプロファイリング結果には使用されていません。

┌─────────────────────────────────────┐
│┌───────────────────────────────────┐│
││              Query Profiling Information              ││
│└───────────────────────────────────┘│
└─────────────────────────────────────┘
explain analyze create or replace table src as select station, min(measurements) min, max(measurements) max, cast(avg(measurements) as decimal(8, 1)) avg from df group by station
┌────────────────────────────────────────────────┐
│┌──────────────────────────────────────────────┐│
││              Total Time: 56.08s              ││
│└──────────────────────────────────────────────┘│
└────────────────────────────────────────────────┘
┌───────────────────────────┐
│           QUERY           │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│      EXPLAIN_ANALYZE      │
│    ────────────────────   │
│           0 Rows          │
│          (0.00s)          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│      CREATE_TABLE_AS      │
│    ────────────────────   │
│           1 Rows          │
│          (0.00s)          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         PROJECTION        │
│    ────────────────────   │
│          station          │
│            min            │
│            max            │
│            avg            │
│                           │
│         8888 Rows         │
│          (0.00s)          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│       HASH_GROUP_BY       │
│    ────────────────────   │
│         Groups: #0        │
│                           │
│        Aggregates:        │
│          min(#1)          │
│          max(#2)          │
│          avg(#3)          │
│                           │
│         8888 Rows         │
│          (121.01s)         │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         PROJECTION        │
│    ────────────────────   │
│          station          │
│        measurements       │
│        measurements       │
│        measurements       │
│                           │
│      1000000000 Rows      │
│          (0.29s)          │
└─────────────┬─────────────┘
┌─────────────┴─────────────┐
│         TABLE_SCAN        │
│    ────────────────────   │
│         Function:         │
│       READ_CSV_AUTO       │
│                           │
│        Projections:       │
│          station          │
│        measurements       │
│                           │
│      1000000000 Rows      │
│         (316.38s)         │
└───────────────────────────┘

パフォーマンス結果

パフォーマンスは、各ライブラリのスタンドアロン スクリプトを作成し、ディスク上の10億行のCSVファイルを指し示すことによって測定されました。

スクリプトは、軽量な手作りのベンチマークツールから実行され、子プロセスが終了するまでpsutilを使用してメモリとCPUメトリクスを50ms間隔でポーリングしました。各ベンチマークでは2回のウォームアップイテレーションを実行し、その後スクリプトを30回繰り返しました。

このアプローチは完璧ではありませんが、比較の目的のために精度とオーバーヘッドのバランスが取れています。

ここでの結果はまさに雄弁です：PolarsとDuckDBはPandasよりも大幅に高速で、それぞれ2倍と19倍少ないメモリを使用します。手作業で作成されたJava実装に迫る勢いです。

Polars実装のメモリ使用量はまだ少し多いように思えます。すべての計算がストリーミングされていなかった可能性があります。DuckDBは完璧で、非常に少ないコードとチューニングなしで驚異的なパフォーマンスを発揮しました。

ローカル開発パフォーマンス

しかし、より良い本番パフォーマンスは方程式の一部にすぎません。DuckDBとPolarsは、ローカル開発ループを高速化するという点でも優れています。

強力でやや古いラップトップハードウェアでパフォーマンステストを繰り返してみましょう。このテストでは、Intel i5-1135G7を搭載し、8コアと16GBのRAMを持つFramework 13を使用しました。

ここでもPolarsとDuckDBから素晴らしいパフォーマンスが見られますが、Polarsはかなりのメモリを消費し、わずかにスワップに手を出しています。対照的に、Pandasはカタツムリのように遅く、メモリをがぶ飲みします。

無料で何が得られるか？

このベンチマークは、従来のPandasワークフローに対するいくつかの重要な利点を浮き彫りにしています：

• 簡単なマルチスレッド処理：PolarsとDuckDBは、スレッドやプロセスを管理する必要なく、すべてのCPUコアを自動的に使用します。それらのコアのために支払ったのですから、使いましょう！
• 効率的なメモリ使用とストリーミング：DuckDBとPolarsはどちらもデータをチャンク単位で処理でき、メモリ使用量を削減するのに役立ちます。
• 遅延評価：計算全体を事前に定義することで、両方のライブラリは実行プランを最適化し、述語プッシュダウン（データを早期にフィルタリングする）などの技術を適用できます。これは「本物の」データベースと同様です。
• ディスクへのスピル（一時退避）の可能性：最適化された操作がRAMを超えた場合、これらのツールには、OSの汎用スワッピングに依存するよりも効率的な、中間結果をインテリジェントにディスクに一時退避する組み込みメカニズムがあります。

より標準的なTPC-Hベンチマーク結果に興味がある場合は、こちらで見つけることができます。

TPC-Hベンチマーク結果に関する注意：これらのTPC-Hベンチマークは、Polars / DuckDBとこの分野で「競合」するホスト型Daskサービスを提供するCoiledによって実行されました。それは彼らのベンチマークが間違っているという意味ではありませんが、ある程度の懐疑心（私自身も含めて！）を維持する価値はあります。

Apache Arrowによる簡単な導入

私たちは皆、派手なツールに以前は打ちのめされてきました。中には、今まさに打ちのめされている人もいるかもしれません。これらのツールをすべて書き直すことなく試すための鍵は、Apache Arrowです。Arrowは、列指向データの事実上のインメモリ表現になりつつあり、Pandasの元の作成者であるWes McKinneyの brainchild です。Pandasは、2023年4月のバージョン2.0リリース以来Arrowをサポートしています。

PolarsとDuckDBはどちらもArrowをネイティブにサポートしています。これは、Polars、Pandas、DuckDB間でメモリをコピーすることなくDataFrameを移動できることを意味し、フレームワーク間の切り替えはほぼ「無料」になります。ここで注意すべき点は、PandasはデフォルトではArrowベースのDataFrameを作成しないことです。DataFrameを作成する際に、dtype_backendをpyarrowにしたいことを指定する必要があります。

動機付けの例：NYCタクシーデータ

2009年以降のNYCタクシーデータ（月ごとのParquetファイルに保存）を分析して、パンデミック（2019-2022年）中に現金払いが少なくなったかどうかを見てみましょう。これには約3GBのParquetデータを処理する必要があります。

この例では、Parquetファイルを読み取る関数と、DuckDBとPandasの両方で計算を実行する関数を実装します。これらの関数のPolarsバージョンは、読者への演習として残しておきます 😉。

Pandasコード：

COLUMNS = ["tpep_pickup_datetime", "payment_type"]
MIN_DATETIME = "2019-01-01"
MAX_DATETIME = "2023-01-01"
# Enum value for cash as the payment type
CASH = 2

def read_data_pandas(folder: Path) -> pd.DataFrame:
    df = None
    for entry in folder.iterdir():
        if not entry.name.endswith("parquet"):
            continue
        temp_df = pd.read_parquet(entry, columns=COLUMNS, dtype_backend="pyarrow")
        if df is not None:
            df = pd.concat([df, temp_df])
        else:
            df = temp_df
    # Clamp the data to the range we're interested in
    df = df[df["tpep_pickup_datetime"] > pd.to_datetime(MIN_DATETIME)]
    df = df[df["tpep_pickup_datetime"] < pd.to_datetime(MAX_DATETIME)]

    df["month"] = df["tpep_pickup_datetime"].dt.month
    df["year"] = df["tpep_pickup_datetime"].dt.year
    df = df.drop(["tpep_pickup_datetime"], axis="columns")

    return df

def calculate_cash_pandas(df: pd.DataFrame):
    df = (
        df.groupby(["year", "month", "payment_type"])
        .agg({"payment_type": "count"})
        .unstack(fill_value=0, level=2)["payment_type"]
        .reset_index()
    )

    df["total_payments"] = df.iloc[:, 2:8].sum(axis=1)
    df["cash_pct"] = (df[CASH] / df["total_payments"]) * 100

DuckDBコード：

def read_data_duck(folder: str):
    return duckdb.sql(f"""
        select 
          datepart('year', tpep_pickup_datetime) year, 
          datepart('month', tpep_pickup_datetime) month, 
          payment_type 
        from '{folder}/*.parquet'
        where 
          tpep_pickup_datetime > '{MIN_DATE}'
          and tpep_pickup_datetime < '{MAX_DATE}'"""
      )

def calculate_cash_duck(data):
    return duckdb.sql(f"""
        with total as (
            select year, month, count(payment_type) payments from df 
            group by year, month
        ),
        total_cash as (
            select year, month, count(payment_type) cash from df 
            where payment_type={CASH} 
            group by year, month
        )
        select total.*, cash, (cash / total) * 100 cash_pct from total 
        join total_cash 
          on total.year=total_cash.year and total.month=total_cash.month
        order by total.year, total.month
    """).df() # Force evaluation by materialising to a df otherwise the execution is lazy

その後、これらの呼び出しを組み合わせて、データと計算を読み込み、どちらのツール（読み込み、書き込み、または両方）を選択するかによって得られる効果を理解できます。

def pure_pandas(folder: Path):
  data = read_data_pandas(folder)
  df = calculate_cash_pandas(data)

def duck_reads_panda_thinks(folder: Path):
  data = read_data_duck(folder).df()
  df = calculate_cash_pandas(data)

def panda_reads_duck_thinks(folder: Path):
  data = read_data_pandas(folder)
  df = calculate_cash_duck(data)

def pure_duck(folder: Path):
    data = read_data_duck(folder)
    df = do_taxi_duck_compute(data)

以前と同じベンチマークスクリプトとラップトップ仕様を使用して、次の結果が得られました：

ここでも同様のパターンが見られます。DuckDBはマシンのCPUコアを最大限に活用しつつ、メモリを少量しか消費しません。Pandasが関与すると、処理速度は著しく低下し、メモリ使用量は急激に増加します。

パンデミック中に現金の使用が減少したかどうかに関心があるなら、確かに減少しました。しかし、相関関係は因果関係ではないため、ここから有意義な結論を導き出さないでください。

［画像: 2019年1月から2023年1月までのNYCでの全タクシー乗車に対する現金使用率のグラフ］

2019年1月から2023年1月までのNYCでの全タクシー乗車に対する現金使用率のグラフ

なぜ私の言うことを聞くべきではないのか？

懐疑主義はテクノロジー業界では健全なことです。そのため、私の言うことを聞くべきではない理由のリストをまとめました：

1. 私は単にベンチマークを実行している人間です。すべてのコードはオープンソースなので、自分で読んで欠陥があるかどうか判断できます。ぜひそうしてください。
2. もしあなたがPandasエコシステムに非常に深く統合されているなら、おそらく切り替えコストが高すぎるでしょう。しかし、「切り替えコストが高すぎる」という基準は、私がこのトークを行ったときからかなり劇的に変化しました。
3. Pandasに「料理」させてください。Pandasは改善されていますが、エコシステムにおけるその中心的な位置を考えると、ゆっくりではありますが。しかし、純粋なパフォーマンスを超えて、DuckDBとPolarsの両方には利点があります。私の経験では、どちらも混乱の少ないAPIを提供しており、DuckDBの場合はSQLは非常に汎用性の高いスキルセットです。

Polars vs DuckDB、どちらが良いか？

これは完全にあなたのワークロード、経験、好みに依存します。データエンジニアはSQLを好み、ソフトウェアエンジニアはPolarsを好む傾向があります。両方を試してみて、どちらが気に入るかを見つけてください。

本当にあなたがすべきでないことは、Pandasがパフォーマンスの悪さに苦しんでいるという理由だけで、盲目的に分散クエリシステムとその付随する複雑さをすべて採用することです。長期的には本当にそれらが必要になる可能性はかなり小さいです。

パフォーマンスはかつてないほどアクセスしやすくなっています。色々試してみてください！