INTRODUCTION
「並列処理を使おう」と最初から決まっていたわけではない
今回の並列処理は、 最初から設計されたものではありません。
もともと引き継いだのは、 1件のデータを投入することを前提に 作られていた既存の移行ツールでした。
大量データを実際に投入してみると、 移行完了までに 約2年半 かかる見込み。
しかも、 仕様書も十分に残っておらず、 詳しい知見者もいない。
まず「なぜこんなに時間がかかるのか」を 調べるところから始まりました。
01
まずCPU使用率を確認した
大量データの処理に時間がかかっている。
そう聞くと、
最初は「Pythonの処理が遅いのかな」と考えます。
そこで、 実際に処理を動かしながら サーバー側のリソースを確認しました。
Linux環境では、
topや
vmstatなどを使って
CPUやメモリ、I/Oの状況を確認しました。
topvmstat 1すると、 処理そのものは動いているものの、 CPUの能力を十分に使えていない 状況が見えてきました。
「この処理、もっとCPUを使えるんじゃないか？」
ここから、 並列処理について調べ始めました。
02
Pythonで並列処理をどう実現するか
Pythonには、 並列処理を実現するための仕組みがあります。
今回調べていった中で使ったのが
concurrent.futuresです。
from concurrent.futures import ProcessPoolExecutor
その中でも、
大量データのCPU処理を複数プロセスで実行するために
ProcessPoolExecutor
を利用しました。
考え方としてはシンプルです。
大量データ
↓
処理単位に分割
↓
複数プロセスへ振り分ける
↓
同時に処理
↓
結果をまとめる
ただ、 実際の実装では 「分割して並列実行する」だけでは終わりませんでした。
03
DataFrameをそのまま処理するのではなく、分割する
大量データを扱う上で、 pandasのDataFrameを利用しました。
import pandas as pd
df = pd.read_csv("data.csv")例えば、 10万件のデータがあるとします。
これを1つの処理として 順番に処理するのではなく、 複数の処理単位に分割します。
100,000件
↓
DataFrame
↓
25,000件 × 4
↓
Process 1
/ Process 2
/ Process 3
/ Process 4
こうすることで、 複数のCPUコアを使って 同時に処理できるようになります。
「大量データをどう分割するか」も 並列処理では重要でした。
04
ProcessPoolExecutorで処理を並列化する
実際の考え方を シンプルなコードにすると、 例えば次のようになります。
from concurrent.futures import ProcessPoolExecutor
def process_data(data):
# データ処理
return result
with ProcessPoolExecutor() as executor:
results = executor.map(
process_data,
data_chunks
)
ポイントは、
process_dataそのものよりも、
その処理を複数プロセスに
振り分けることです。
これまで、
データ1
→ 処理
データ2
→ 処理
データ3
→ 処理
データ4
→ 処理
と順番に処理していたものを、
データ1 ─┐
データ2 ─┼→ 並列処理
データ3 ─┤
データ4 ─┘
という形に変えていきました。
05
でも、プロセスを増やせば増やすほど速いわけではない
ここが、 実際にやってみて難しかったところです。
「CPUを使えていないなら、
プロセスをもっと増やせばいい」
と単純に考えたくなります。
しかし、 実際にはそう簡単ではありません。
CPUだけを見て 並列度を上げてしまうと、 メモリ使用量が増えたり、 I/Oがボトルネックになったりします。
さらに、 GIS側の処理速度が限界なら、 Python側だけ高速化しても 全体としては速くなりません。
「CPUを100%にすること」が目的ではなく、 「全体として一番速くなるポイント」を探す。
06
並列度は実際に動かして探した
最初から 「4プロセスが最適」 「8プロセスが最適」 と分かっていたわけではありません。
実際に処理を動かして、 結果を確認して、 また条件を変えてテストする。
並列数を設定
↓
実際に処理
↓
処理時間を計測
↓
CPU使用率を確認
↓
メモリ・I/Oを確認
↓
結果を比較
↓
並列数を調整
こうしたテストを繰り返しながら、 どこまでリソースを使っても システム全体に無理がないのかを探しました。
07
「CPUをギリギリまで使う」という感覚
この案件で印象に残っているのが、 CPUの使用率を見ながら 処理を調整していったことです。
もともとの処理では、 サーバーのCPUにまだ余裕がある。
それなら、 その余っているリソースを 移行処理に使えないか。
「せっかく使えるCPUがあるなら、 できるだけ使ってしまおう」
ただし、 本当にギリギリまで使う場合には、 システムへの影響も考える必要があります。
特に、 移行処理だけを実行する環境と、 他の処理も動いている環境では 最適な設定が変わります。
そのため、 「最大性能」と 「運用しながら安全に動かせる性能」 を分けて考えるようになりました。
08
運用しながらなら、あえてCPUを余らせる
移行処理だけを実行するのであれば、 できるだけリソースを使って 処理時間を短くできます。
しかし、 実際のシステムでは 移行処理だけが動いているとは限りません。
他の処理も動いている環境で CPUを完全に占有してしまえば、 当然ながら別の処理へ影響が出ます。
実際の案件でも、 「全部使えば約2週間」 という状態と、 「運用しながらなら約2か月」 という状態を 分けて考えました。
性能テストで重要なのは、 最高速度だけではありません。
実際の運用条件で、
どこまで高速化できるのか。
そこまで考えて初めて、
実用的な性能改善になります。
09
並列処理で一番難しかったのは「分け方」
並列処理というと、
ProcessPoolExecutor
を使うこと自体が
難しいように見えます。
しかし、 実際に考えることが多かったのは、 「何をどう分割するか」でした。
データを細かく分けすぎると、 プロセス間のやり取りや データの受け渡しが増えます。
逆に、 分割が大きすぎると、 CPUを十分に使えません。
小さすぎる
→ 管理コストが増える
大きすぎる
→ 並列化の効果が小さい
ちょうどよいサイズ
→ CPUを効率よく使える
つまり、
並列処理は
「何個に分けるか」だけではなく、
「どの単位で仕事を分けるか」
が重要でした。
FIELD NOTE
最初から「並列処理を設計した」わけではない
この案件を後から説明すると、
大量データをDataFrameに格納し、 データを分割。
ProcessPoolExecutorで並列処理し、 CPUリソースを有効活用した。
と書けます。
もちろん、 技術的にはその通りです。
でも、 実際の進め方は そんなにきれいではありませんでした。
「CPUが余っている」
そこから、
何でCPUを使えていない？
↓
並列処理できないか？
↓
Pythonでどうやる？
↓
ProcessPoolExecutorを試す
↓
CPU使用率が上がった
↓
でもメモリは大丈夫か？
↓
I/Oはボトルネックになっていないか？
↓
またテストする
そんな感じでした。
最初から完成形があったのではなく、 テストするたびに次に調べることが見つかった。
そして、 それを繰り返しているうちに 並列処理の仕組みが出来上がっていきました。
10
並列化したら終わり、ではなかった
並列処理を入れたことで、 処理速度は大きく改善しました。
しかし、 そこで終わりではありません。
並列化によって、 今まで見えていなかった 別のボトルネックが見えてくることがあります。
並列化する
↓
CPU使用率が上がる
↓
別の処理がボトルネックになる
↓
そこを調べる
↓
改修する
↓
再度性能テスト
つまり、 並列処理は 「最後の魔法」ではありません。
ボトルネックを次の場所へ移動させる技術でもあります。
だからこそ、 並列化した後の計測が重要でした。
11
最終的に、2年半という数字を現実的なところまで縮めた
並列処理だけで すべてが解決したわけではありません。
pandasによるデータ処理、 データ分割、 ProcessPoolExecutorによる並列処理、 CSV読み込み、 Linuxのリソース確認、 性能テスト。
それぞれを少しずつ改善していきました。
当初の見込み
約2年半
↓
処理改善・並列化・性能調整
↓
約2週間
※移行処理にリソースを集中できる場合
また、 実際の運用と並行して移行する場合には、 リソースをすべて使うことができません。
その場合でも、 約2か月 で移行完了を見込めるところまで 持っていくことができました。
SUMMARY
並列処理から学んだこと
今回の案件で、 単純に 「ProcessPoolExecutorを覚えた」 というだけではありません。
そして、 一番大きかったのは、
「処理が遅い」という結果だけを見て終わらず、 どこで時間を使っているのかを調べること。
CPUが余っているなら、 なぜ余っているのかを調べる。
並列化したら、 次にどこがボトルネックになったのかを見る。
メモリやI/Oが問題なら、 そこを見る。
そして、 またテストする。
並列処理は、 「速くするためのコード」ではなく、 「システム全体を見ながら処理能力を引き出すための手段」 だった。
既存ツールを引き継ぐ
↓
大量データで性能問題が発覚
↓
CPU・メモリ・I/Oを確認
↓
pandas / DataFrameでデータ処理
↓
データを分割
↓
ProcessPoolExecutorで並列化
↓
性能テスト
↓
ボトルネックを再確認
↓
再改修
↓
現実的な移行時間へ
最初から 「こういう構成にすればいい」 と分かっていたわけではありません。
ただ、 CPU使用率を見て、 コードを調べて、 試して、 測って、 また直す。
そうやって一つずつ試していった結果、 並列処理を使った大量データ移行の仕組みが 出来上がっていました。
これが、 この案件で経験した Pythonによる並列処理でした。