Parquetで時系列データを扱う

こんにちは、ギークプラスの中村です。
この記事では「大量の追記データを安く・速く・壊れずに扱う」ために Parquet を選んだ理由を、需要予測のプロトタイプの事例で紹介します。
本記事ではバックエンドエンジニア・データエンジニア・MLエンジニアなどを主な対象に、S3＋Parquetを前提としたスケーラブルなデータ基盤・ストレージ設計に関心のある方向けに書いています。

---

背景

時系列データは、需要予測・在庫計画・ダッシュボードなど、サプライチェーン領域のあらゆるロジックの基盤となる存在です。当社でも、需要予測エンジンのプロトタイプを構築する中で、「どのフォーマットで時系列データを保存すべきか」 は初期から重要でした。

一見すると選択肢は多く、CSV・JSON・RDB・NoSQLDB・各種バイナリフォーマット・Data Lake系フォーマット（Parquet、ORC、Avro）などが候補になります。しかし検討の結果、私たちは Parquet を時系列データの中心フォーマットとして採用する方向が明確になりました。

この記事では、その理由や実際につまずいたポイント、ビジネス要件からくる特有の考慮点、そしてこれから拡張していく上で考慮する点までをまとめます。

時系列データのユースケースの特徴

私たちのビジネス要件における時系列データには、次のような特徴があります。

データ量が “永遠に増え続ける”

組織 × 商品 × 倉庫 × 日付 で行数は際限なく増加します。
例：10万 SKU(Stock Keeping Unit) × 10倉庫 × 365日 = 年間3650万行。

履歴とバージョン管理が欠かせない

需要予測は「いつ生成した予測か」に意味があります。
監査性・再現性の観点から、過去のファイルを安易に上書きできません。

クエリは列単位での抽出が多い

読み込み時は「組織、SKU、と倉庫を絞って、特定期間の値だけ見たい」というケースが大半です。

書き込みはバルク、読み込みはインタラクティブ

予測生成はバッチで大量に出力し、利用者はWeb UIで即座に確認したい。

コストはビジネススケールに直結する

長期的に顧客数が増える前提のため、データレイク基盤は必ず“低コスト”である必要があります。

これらの背景から、データは 「append-onlyで、圧縮効率が良く、列指向で、バージョン管理と相性が良い」 形式が理想となります。
そこで最有力候補になったのが Apache Parquet でした。

なぜデータベースではなく S3（ブロブストレージ）に時系列データを保存する決定をしたのか？

RDB / NoSQLDB / 時系列DB はそれぞれ、トランザクション処理やポイントルックアップ、短期間の高頻度メトリクス集計には強みがあります。
一方、当社のように 「マルチテナントの年単位で増え続ける履歴時系列を、バージョンを変えながら長期保管し、後から自由に分析・再学習したり、DataFrame に読み込んで学習パイプラインを回す」 ユースケースでは、

• ストレージ単価とバックアップコストが高くなりやすい
• テーブル/パーティション/インデックス設計が複雑化しがち（特に NoSQLDB・時系列DB）
• 1社ごとのスキーマやアクセスパターンの変化に、運用コストなしで追従しづらい

といった理由から、「単一のDBに時系列データを全部載せる」戦略はスケールしないと判断しました。
そこで、長期保存するSingle Source of Truth (SSoT)は S3 + Parquet に置くのがベストだと判断しました。

他形式と比較した時のParquetの強み

CSV / JSON は扱いやすいが、スケールしない

• 圧縮が弱い
• 列単位読み込みができない
• スキーマ保証がない
• 巨大ファイルは読み込みが重い
  など

検証中、CSVでのクエリはシンプルですが 数百万行を超えた瞬間に実運用の限界が見える という実感がありました。

Parquetのメリット

1. ファイル圧縮率が高い

以下は簡単なスクリプトで手元のCSVを複数フォーマットに変換した時のデータサイズの比較です。もちろん、本図の結果は当社のデータおよび変換スクリプトに基づく一例であり、すべてのケースで同様のサイズ比になるとは限りません。

2. 列指向のフォーマット

私たちのユースケースでは
「商品A の過去2年の需要だけ欲しい」
「特定の倉庫だけフィルタしたい」
「予測値と実績をJOINしてWMAPEを算出したい」
など、列ベースのフィルターを行うクエリが多発することを考慮すると列指向フォーマットであるParquetを使うことで、クエリの速度が良くなることが期待できます。

Parquetの構造

行グループ→列チャンク→ページという構造になっています。特定の列の値だけクエリで見たい時は、各行グループの特定の列チャンクだけ参照すれば良いので、行全体を読み込む必要がありません。

Parquet File
 ├── Row Group 1
 │     ├── Column Chunk (col1)
 │     │       ├── Page 1
 │     │       ├── Page 2
 │     │       └── Page N
 │     ├── Column Chunk (col2)
 │     └── ...
 ├── Row Group 2
 │     ├── Column Chunk (col1)
 │     └── ...
 └── ...

Parquet公式Doc
概念の説明：https://parquet.apache.org/docs/concepts/
概念の上下関係：https://parquet.apache.org/docs/file-format/metadata/

3. 利用ツールとの互換性

私たちが利用するPythonのライブラリのPandasやAWSのAthenaなどのツール群に楽に導入できることも良い点です。

Parquetのデメリット

1. 対人可読（Human readable）ではない

Parquetに対応しているアプリは少ないので中身を見ようとした時にはCSVのように扱いやすくはないです。個人的にはVSCodeではParquet Explorer、ターミナルではTabiewを使って中身を見てます。

2. イミュータブル

バージョン管理を行い、基本append-onlyで考えている私たちのユースケースでは問題にはなりませんが、変更をしたい場合は、ファイル全体を再生成する必要があります。

3. 行単位のアクセス

一つの行のレコードだけ取得したいユースケースでは、列指向のParquetよりも行指向のファイルフォーマットの方が適切と考えられます。

需要予測プロトタイプ開発で直面した課題

プロトタイプでは、Athenaをクエリしてバックエンドが時系列データを取得するのではなく、バックエンドでparquetファイルをダウンロードしてキャッシュして保持して、直接バックエンドからparquetファイルを読み込んで指定の組織 × 商品 × 倉庫のデータを取得する実装にしました。
そこでなぜかバックエンドのAPIのレイテンシーが悪化している問題に直面しました。
原因を調査したところ、ParquetのデータをPythonのlistに読み込んでフィルター処理する部分が時間がかかっていました。そこで、C++実装を使っていてフィルター処理を行うPythonのApache ArrowライブラリのPyArrowで処理することでレイテンシを大幅に改善することができました。(3~5s -> 100~300ms)
しかし、これはあくまでプロトタイプでの実装の対応で、本番ではよりシンプルな実装でAthena Icebergテーブルにフィルターロジックの責務を任せることを想定しています。

---

実際の実装を簡略化したpyarrowを使ったフィルターの例:

import pyarrow as pa
import pyarrow.compute as pc

def _filter_table_by_item(
    table: pa.Table,
    *,
    item_column: str,
    sku_id: str,
    warehouse: str,
) -> tuple[pa.Table, bool]:
    """Return filtered table and whether Arrow fast-path was used.

    Arrow compute keeps the filtering in C++ (no Python to_pylist or loops),
    which is orders of magnitude faster on 1M+ rows and avoids GC pressure.
    We fall back to the legacy Python path only if the vectorized masks find
    no matches, to preserve compatibility with edge-case item encodings."""
    item_col = _column(table, item_column)
    mask = pc.equal(item_col, _compose_item_key(sku_id, warehouse))
    filtered = table.filter(mask)
    if len(filtered):
        return filtered, True
    return filtered, False

まとめ

私たちの時系列データのユースケースになぜParquetが適切だと判断して、採用するに至った流れをざっくり説明しました。時系列データというのは結構広い括りではあるので、皆さんのアプリのユースケースによっては時系列データベースを利用することや別のファイルフォーマットでS3上に管理することが望ましいと判断できるケースは多いと思います。