Posted inプログラミング

DuckDBとPythonによるローカルデータ分析：データベースサーバーなしで数千万行をクエリする

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

現場のリアルな課題：5,000万行のデータでPandasが破綻するとき

先月、私たちのチームはよくあるボトルネックに直面しました。48個のCSVファイルおよびParquetファイルに分散した、月間35 GBの生トランザクションログを解析する必要があったのです。目的はシンプルで、要約統計量の集計、ユーザーのコホート分類、そして異常なクレジットカード決済のフラグ立てでした。

当然のように、私たちは最初にPandasを採用しました。しかし、それが間違いでした。わずか5.2 GBのファイル1つに対してpd.read_csv()を実行しただけで、40秒以内に実メモリ使用量が16 GBを突破。ワークステーションはフリーズし、スワップ領域が激しく消費され、最終的にLinuxのOOM（Out of Memory）キラーによってPythonプロセスが読み込み途中で強制終了されました。そこでchunksize=100_000を指定して入力を分割処理しようと試みました。クラッシュこそ回避できたものの、チャンクをまたいだユニークユーザー数のカウントなど、状態を保持する集計を行うには、何百行もの脆弱なグルーコードが必要になりました。

代替案として、DockerでローカルにPostgresやClickHouseのコンテナを立ち上げることも検討しました。しかし、分析用データベースのプロビジョニングには別の煩わしさが伴います。厳格なDDLスキーマを定義し、共有バッファをチューニングし、ホストポートをマッピングした上で、クエリを1本実行する前のデータ取り込みだけで20分も待たされることになります。

根本原因の分析：メモリフットプリントと行指向の限界

データセットが数ギガバイトを超えると、なぜ標準的なデータ処理ツールは苦戦するのでしょうか？その原因は、Pythonのデータ構造と行指向エンジンの内部動作にあります。

  • Pythonオブジェクトのオーバーヘッド：標準のPandasは、明示的にPyArrowエンジンを設定しない限り、文字列を汎用的なPythonオブジェクトのポインタとして保持します。各セルには40〜60バイトのメタデータオーバーヘッドが付加されるため、1.2 GBのCSVファイルが物理RAM上で5.8 GBにまで容易に膨れ上がります。
  • 防衛的コピー（Defensive Copying）：df[df['status'] == 'completed']のようなフィルタリング式は、デフォルトで完全に新しいDataFrameをメモリ上に確保します。ビューの管理やガベージコレクションを手動で行わない限り、3つの操作をチェーンするだけでメモリ消費量は3倍に跳ね上がります。
  • 行指向 vs カラム型（列指向）処理：SQLiteは行データを連続して格納します。4,000万件のレコードに対してAVG(latency)を計算する場合、SQLiteはたった1つの整数値を取り出すためだけに、すべてのタイムスタンプ、IPアドレス、ペイロードをディスクから読み出します。これによりディスクI/Oが無駄に消費され、不要な属性でCPUキャッシュが汚染されます。
  • シングルスレッドのボトルネック：SQLiteもコアなPandasも、分析処理を単一コアで実行します。1つのスレッドが使用率100%に張り付いている間、CPUの残りの7コアや15コアは完全にアイドル状態のままです。

ソリューションの比較：Pandas、SQLite、Spark、DuckDB

開発者向けのマシンで1,000万〜1億行のデータを分析する場合、主に4つの選択肢が挙げられます。

1. Pandas / チャンク処理

データがシステムRAMの20%前後に無理なく収まる場合、Pandasは素晴らしい性能を発揮します。しかしその制限を超えると、実行プランナーが存在しない点が痛手となります。チャンク分割のロジックは脆弱で、単純な集計作業すら保守性の低い悪夢へと変えてしまいます。

2. SQLite

SQLiteはトランザクションを伴うアプリケーションストレージとしては極めて堅牢です。しかし、並列スキャンやカラム型ストレージを備えていないため、5,000万件のレコードに対するグループ化集計には数秒ではなく数分かかります。

3. ローカルのApache Spark / PySpark

Sparkは大規模なワークロードを難なく処理できます。しかし、単一マシン上でローカル実行するには大がかりすぎます。探索的なスクリプトを動かすだけのために、JVMのオーバーヘッド、15秒かかるエグゼキュータの起動時間、難解なメモリチューニングフラグと格闘することになります。

4. DuckDB

DuckDBはこのギャップを埋める存在です。「分析用途に特化したSQLite」と考えるとわかりやすいでしょう。バックグラウンドサービスを一切必要とせず、組み込み型C++エンジンとしてインプロセスで動作します。内部ではベクトル化カラム型実行エンジンを採用し、すべてのCPUコアをフル活用しながら、RAM容量を超えるデータセットでもディスクから直接ストリーミング処理します。

最適なアプローチ：DuckDBとPythonによる実践的な実装

最も洗練された構成は、DuckDBのベクトル化SQLエンジンをPythonと直接組み合わせることです。ディスク上のファイルをインプレースでスキャンし、C++層でデータを処理して、最終的にグループ化された結果のみをPandasまたはPolarsのDataFrameとして実体化します。

1. インストール

DuckDBは外部依存関係なしのビルド済みホイールを提供しています。pip経由でスタックをインストールします。

pip install duckdb pyarrow pandas

2. データベースへのインポートなしでRAWファイルを直接クエリする

DuckDBはヘッダーを読み取り、ファイルを直接クエリできます。テーブル定義や手動でのデータインポート手順は一切不要です。

import duckdb

# 分割されたParquetファイル全体の2,500万件のレコードをクエリ
query = """
    SELECT 
        merchant_id,
        COUNT(*) AS total_transactions,
        ROUND(AVG(amount), 2) AS avg_amount,
        ROUND(SUM(amount), 2) AS total_revenue
    FROM 'data/transactions_*.parquet'
    WHERE status = 'SUCCESS'
    GROUP BY merchant_id
    HAVING COUNT(*) > 1000
    ORDER BY total_revenue DESC
    LIMIT 10;
"""

# DuckDBがParquetのフッターをストリーミングし、約0.8秒で結果を返す
results = duckdb.sql(query)
results.show()

3. Python DataFrameとのゼロコピー連携

DuckDBはメモリバッファをApache ArrowやPandasと直接共有します。負荷の高い計算処理を自身のエンジン側へプッシュダウンし、シリアライズのオーバーヘッドなしでスリムな集計行のみを返します。

import duckdb

con = duckdb.connect(database=":memory:")

# 4,000万行を500行に集約し、Pandasにエクスポート
summary_df = con.execute("""
    SELECT 
        DATE_TRUNC('day', timestamp) AS tx_date,
        country_code,
        COUNT(DISTINCT user_id) AS active_users,
        SUM(amount) AS gross_volume
    FROM 'data/raw_access_logs.parquet'
    GROUP BY 1, 2
""").df()

print(f"処理済みサマリーの形状: {summary_df.shape}")
print(summary_df.head())

4. RAM容量を超えるデータセットの処理（アウトオブコア処理）

16 GBのノートPCで80 GBのクリックストリームダンプをクエリする必要がある場合はどうでしょうか？DuckDBはディスクへのメモリスピル（退避）を自動的に処理します。メモリ消費量の上限を明示的に指定し、一時ファイルを高速なNVMeストレージに向けることが可能です。

import duckdb

# RAM使用量の上限を設定し、ディスクスピルを構成
con = duckdb.connect(database="analytics_cache.duckdb")

con.execute("SET max_memory = '10GB';")
con.execute("SET temp_directory = '/tmp/duckdb_spill';")
con.execute("SET threads = 8;")

# アウトオブコア処理を実行し、圧縮Parquetとしてディスクに保存
con.execute("""
    COPY (
        SELECT 
            user_id,
            COUNT(*) AS event_count,
            APPROX_COUNT_DISTINCT(session_id) AS approx_sessions
        FROM 'data/huge_clickstream_dump.csv'
        GROUP BY user_id
    ) TO 'output/user_behavior_metrics.parquet' (FORMAT PARQUET, COMPRESSION ZSTD);
""")

con.close()

現場で培った実践的なノウハウ

日々のETLスクリプトや定期実行ジョブでDuckDBを運用する中で得られた、3つの重要な教訓を紹介します。

  • CSVよりもParquetを優先する：DuckDBは並列スニファーによってCSVを高速にパースします。しかし、Parquetにはカラム統計量とバイトオフセットが含まれています。これにより、DuckDBはWHEREフィルタを評価する際に不要なファイルブロック全体をスキップできるため、20 GBのデータセットに対するクエリ時間が42秒から3.5秒へと大幅に短縮されます。
  • インメモリのPandas DataFrameを直接クエリする：DuckDBはPythonスコープ内のアクティブなDataFrameを、インメモリSQLテーブルのように扱えます。データ変換の手間は不要で、duckdb.sql("SELECT * FROM my_df WHERE score > 90")と書くだけですぐに動作します。
  • ファイルロックの競合に注意する：DuckDBファイルは複数のリーダーによる同時読み込みをサポートしていますが、ライターは同時に1つしか許可されません。ワークフローで並列ワーカープロセスを実行する場合は、各ワーカーに独立した一時ファイルへ書き込ませた後、統合ビューを介して集約してください。

このワークフローの効果は絶大です。私たちの35 GBのパイプラインでは、手動のチャンク処理スクリプトと使い捨てのデータベースコンテナをDuckDBに置き換えたことで、パイプラインの実行時間が18分から46秒に短縮されました。コードベースは短く堅牢になり、インフラの維持管理コストもゼロになりました。

Share:
Tags: