設計した実行ルール

Dagster、Trino、DataFusion、PostgreSQL を使う ELT 基盤のメモリ割り当てを設計しました。前提は AMD EPYC 9175F 16c/32t、768GB DDR5-6400、RTX 6000 Pro Max-Q、NVMe 3.84TB です。このうち 512GB を ELT 専用プールとし、ステップごとに再利用する方針です。

Airbyte -> Iceberg / Trino -> 可視化 をローカル・オフライン中心で動かすデータ基盤を想定しています。48h Prototype や データ基盤整備 の提供形につなぐ構想の、実行ルールを整理したものです。

メモリのピークを分ける

768GB あっても、Trino の heavy query、DataFusion の変換、PostgreSQL のロード、Dagster の並列 worker が重なると、メモリのピークが衝突します。swap や OOM に加え、途中で失敗した処理の再開も考える必要があります。

メモリの使い方を次のように決めました。

512GB をステップごとに再利用します。軽い処理は In-Memory、重い処理は Arrow IPC または Parquet に退避します。

この方針で、処理速度と失敗時の再開方法を一緒に設計します。

基本リソース前提

設計の前提です。

  • サーバー: AMD EPYC 9175F 16c/32t (5GHz) + 768GB DDR5-6400 + RTX 6000 Pro Max-Q + NVMe 3.84TB
  • ELT プール: 512GB をワークフロー専用に確保
  • 帯域: メモリ帯域 ~600GB/s、NVMe I/O 数 GB/s、ネットワーク 10GbE

512GB は、実行中の ELT ステップに割り当てる資源です。取り込み、可視化、MLOps のサービスが増えても、ワークロードごとに上限を管理します。

メモリ運用ポリシー

原則

メモリの原則は次の3点です。

  • 512GB プールをステップごとに循環利用する
  • 同一プロセスなら In-Memory 直渡しを優先する
  • 別プロセスやコンテナに分ける場合は、合算で 512GB 上限に収める

重い分析処理は1本ずつ実行し、終了後に解放して次のステップに渡す方針です。Dagster の op を増やす際も、この制限を守ります。

切替基準

中間データのサイズで方式を切り替えます。

中間サイズ推奨方式備考
<= 40-50% (200-250GB)In-Memory（同一プロセス + InMemoryIOManager）帯域 600GB/s を直接使えるので最速
50-80%Arrow IPC ストリームNVMe 退避でもオーバーヘッドを抑えやすい
>80% または複数 op 同時Parquet 一時出力失敗時再開や安全性を優先

この基準に沿って、メモリ内で渡すか Parquet に保存するかを決めます。

各コンポーネントの最適化

DataFusion (Rust)

DataFusion では巨大な中間データの collect を避けます。

  • MemoryPool 上限を op 単位で S/M/L プロファイルとして設定
  • DiskManager は NVMe の /opt/tmp を指定し、必要なら強制スピル可能にする
  • Streaming API の execute_stream() を優先する
  • パーティション数は実コア数である 16c に合わせる

ストリームで処理できる部分を集めずに流し、メモリのピークを抑えます。

Trino

Trino は設定と実行本数の両方を制限します。

  JVM: -Xmx192g
max-spill-per-node=512GB
query.max-memory-per-node=128GB
query.max-total-memory-per-node=160GB

コンテナは 256GB とし、NVMe に spill します。heavy query は Dagster の concurrency 制御で同時 1 本に制限します。

PostgreSQL

PostgreSQL は永続化と大量ロードを担当します。

  shared_buffers=8-16GB
work_mem=128MB
maintenance_work_mem=2-4GB
wal_compression=lz4
checkpoint_timeout=20-30min
max_wal_size=12-16GB

大量ロードには、次の方式を使う方針です。

  • UNLOGGED TABLE -> COPY -> インデックス作成 -> LOGGED
  • CTAS (CREATE TABLE AS SELECT)

必要に応じて取り込み中だけ synchronous_commit=off を使います。常時無効にはしません。

Dagster

Dagster の executor を処理の重さで切り替えます。

  • 軽量: in_process_executor + InMemoryIOManager
  • 重量: multiprocess_executor + opごとに Podman コンテナ (--memory/--cpuset-cpus)

タグで並列数を制限します。

  • heavy 系 (Trino / DataFusion 大 op): max_concurrent=1
  • 軽い op（メタ処理、dbt run、ログ）: 並列可

異なる job でも同じ実行方針を使います。

プロセス / コンテナ設計

In-Memory（最速）

同一ジョブを in-process executor で動かし、pyarrow.Table や RecordBatch を shallow copy で渡す方式です。中間サイズ <=200GB 程度、ELT プールの 50% 以下を条件にしています。

この条件を超える場合は、後続 op を起動するときのメモリ増加も考慮します。

opごとのコンテナ分離

各 op を podman run --memory=... --cpuset-cpus=... で分離します。終了時にプロセスのメモリが解放され、次のステップが 512GB のプールを再利用できます。

Podman を使ったローカル環境の再現にも、この方式を使う方針です。

子プロセス起動前のメモリ解放

Python の親が大きいバッファを持ったまま Rust 子プロセスを起動すると、合計メモリが増えます。起動前に次を行います。

  • 子起動前に親のバッファを del
  • スコープを抜ける
  • gc.collect() を呼ぶ
  • Python は fork より spawn を推奨する

fork の Copy-on-Write が破れると、実メモリの使用量が増えるためです。

I/O 最適化

中間データの形式と置き場を次のように分けます。

  • Arrow IPC: 速度を優先する退避先。ゼロコピーに近い受け渡し
  • Parquet: 安全性とリトライ性重視。圧縮は zstd(1-3)、行グループは 128-256MB
  • NVMe: スピルディレクトリを /opt/tmp に集約
  • /dev/shm: Arrow Plasma / IPC なら --shm-size 2-4GB
  • クリーンアップ: ジョブ終了時に一時ファイルを削除し、失敗時もハンドルする

Arrow IPC は速度、Parquet は保存とリトライを重視する場合に使います。

運転ルール

運用ルールは次の5点です。

  1. 重いクエリは 1 本ずつ実行します
  2. op ごとに終了させて解放し、512GB を循環利用する
  3. 中間サイズに応じて In-Mem -> Arrow IPC -> Parquet を切り替える
  4. 障害許容性が必要な処理は Parquet を選ぶ
  5. 速度最優先の処理は In-Memory で shallow copy / ポインタ渡しを使う

複数の ELT job を開発する際も、同じ基準で実行方式を選びます。

期待する効果

この設計で期待する効果です。

  • 速度: In-Memory でメモリ帯域 600GB/s を利用する
  • 安定性: 512GB の再利用、cgroup 制御、spill でメモリ不足を抑える
  • 柔軟性: 中間サイズで Arrow IPC / Parquet を切り替える
  • 実行管理: Dagster の executor で方式を変える

これらは設計上の狙いです。単機での同時実行を制限し、メモリのピークを管理します。

まとめ

512GB を再利用し、軽い処理は In-Memory、重い処理は Arrow / Parquet に退避する方針です。コンテナを終了させて op のメモリを解放します。

設定への反映予定

次は podman-compose.yml のメモリ / CPU プロファイルと Dagster executor の設定に反映する予定です。Airbyte OSS、Great Expectations、OpenMetadata、MLflow を含む周辺サービスも、同じ実行ルールに沿って構成する計画です。