データ変更を伴うバッチ処理を書く時に考慮していること

こんにちは、id:shallow1729です。最近はインフラ寄りなお仕事をよくやっていますがこれまでにいくつかデータ移行やデータ基盤構築などのバッチ処理のお仕事をしてきました。以前にも一度そういった経験を元に記事を書いたのですが、MySQLやシステムに関する知識が以前よりも増えた今もう一度書き直したいなと思いました。

なので今回はバッチ処理を書く時のテクニック2022版という感じです。今の仕事の関係でMySQLやrailsを前提にしている話が多いですが、おそらく他のデータベースを使っている人にも役に立つ話が多いのではないかと思います。ただ、今回の記事は経験に基づくものが多く、あまりよくないアイデアもあるかもしれません。改善点や間違いなどあればご指摘ください。

冪等性を持つように

冪等性とは端的に言えばある操作を複数回実行しても一回しか実行しなかった時と同じ結果になる性質の事です。長時間かかるバッチはしばしば途中で失敗します。事前の試運転で気づけるバグもありますが、デッドロックやロック待ちのタイムアウトのような並列実行の影響やネットワークタイムアウト、サーバーのOOM kill、外部apiのrate limitなど実際に動かしてみて初めて気づく事もよくあると思います。なのでこけたらもう一回実行できるように作る事が重要です。

冪等性を壊すやつらとしては例えば以下のような奴らがいます。

• auto increment
• uuidやulidなどのランダムな値
• タイムスタンプ

もちろん全てのデータが冪等である必要は無いと思います。例えば更新のタイムスタンプとかは都度最新のものになって欲しい事があると思います。

ただ、例えばあるテーブルのレコードの内容を別のテーブルにコピーする時にコピー先のidを新規に払い出したりすると元のテーブルでどこまでコピーされたかを把握するのは難しくなります。なので基本的にはなるべく先ほど述べたような冪等性を壊すようなやつらは新規に払い出さない方がいいです。

とはいえ。例えば水平シャードされていてidがグローバルにユニークじゃないケースでそれらをマージしたものを作ろうとするとこういった問題は避けられません。そういったケースでうまくやる作戦の一つとして「これらの値を払い出すフェーズとバッチ処理のフェーズを分割しておく」という手はあると思います。例えばあるテーブルのデータを別のテーブルにコピーする時はコピー元とコピー先のidのマッピングを先に作って永続化しておけば、再実行するときにinsert済の箇所はすでにレコードが入っている事が分かります。こうなっていればupsertにすれば冪等にできるかなと思います。

途中からの再実行をできるように

短時間のスクリプトであれば冪等性を担保できれば十分ですが、数時間かかる処理の場合は冪等性に加えて途中からの再実行も欲しい機能です。丸一日かけて90%まで進んだところで失敗して1からやり直しはつらい事が多いと思います。また、「普段は大丈夫だったけどある日突然大量のインプットが来て何回再実行してもタイムアウトで死ぬ」という事はあるので大事なバッチ処理は先んじて再実行できるようにしておけると良いと思います。

途中からの再実行を行うためには「処理をこまめにセーブする事」と「まだ処理がされていない箇所を把握できる事」が特に大事だと思います。「処理をこまめにセーブする」にはDBへのリクエストならトランザクションを細かく分割した上で、何かしらの形で「ここまでは処理を終えた」と分かるようにする必要があります。重要なのは「ここまでは終えた」と分かるように処理を作る必要があるという事で、データ更新ならid何番までは処理した、とか何時より前のデータは処理したと分かるように作る必要があるという感じです。

「途中から再実行」の技術はページネーションで調べるとオフセット法やシーク法というのが出てくると思います。MySQL的には一般にシーク法の方が嬉しいですがオフセット法の方が書きやすい印象なので無理に最初から最適化しなくていいと思います。

あわせてですが、途中からの再実行を実現するためには「途中の状態が許容される事」も重要です。稼働中のサービスの場合に一部のデータは更新済み、他は未更新という状態が許容されるようにバッチ以外含めて設計しておけると良いと思います。

ユーザー影響の削減

稼働中のサービスの裏でバッチが動く時はユーザー影響をなるべく小さくなるように作った方が良いです。

トランザクションはなるべく短く

トランザクションはなるべくすぐに抜けるようにした方が良いです。長いトランザクションはシステム全体のパフォーマンスを悪くしますし、この後説明するギャップロックのように想定外にユーザー影響を与えるロックを取ってしまうリスクがあるのでリスクヘッジの意味でもトランザクションを短くした方が良いです。トランザクションを開始する前にできる事をなるべくやってから開始、最低限の処理をやってトランザクションをコミットするのが大切です。

また、ユーザーもアクセスするようなテーブルについて大量のデータを一度に更新するのは避けて、ほとほどのサイズに分けて更新すると良いです。細かすぎるとN+1問題の弊害が大きくなるのでほどほどが大事です。 バッチスクリプトのパフォーマンスを確認する時はトランザクションの開始から終了までの時間を計ると良いです。この時間がユーザーがそのデータの書き込みできない時間になります。その時間がサービス的に許容できるか考えるのが大事です。

ロックをとる範囲はなるべく少なく

「トランザクションはなるべく短く」に近いですが、ロックを取る範囲はなるべく少なくすると良いです。例えば今のデータを読んでその結果を元に何か更新するような場合は最初のread時にロックを取って、更新してから手放すのが基本ですが、データが変更される心配が無いケースであったり、データの正しさが結果整合で十分な場合などはread時にlockを取らないという戦略もあります。

MySQLの場合ロックを取る範囲を確認したい時はInnoDB Lock Monitorが便利です。これはinnodb_status_output_locksを有効にすると確認できます。例えばupdate … where cid=1;(cidはなんらかの外部キー)みたいなケースではこの検索に使うsecondary indexの検索中に操作したレコードのロックと後述するギャップロック、更新対象のレコードのprimary keyのレコードロックが確認できると思います(REPETABLE READの場合)。また、適切なsecondary indexがない場合使われるindex上で走査されたレコードが全てロックを取られると思います。

可能な限り削除/更新対象はユニークに定まるようにする

ロックについて特に見落としがちなのがMySQLでデフォルトのREPEATABLE READの場合に発生するギャップロックの影響です。例えばupdate … where cid=1;みたいなクエリを考えます。この時cidはnon uniqueな外部キーのカラムで、cidのインデックスが貼られているとします。このクエリが実行中ロックを取る範囲は手元で試したところ「cidのindexのcid=1のレコードロック」、「cidのindexのcid=1の値が入る場所(新規にcid=1のレコードがinsertされる場所)のギャップロック」、「primary keyのcid=1のレコードロック」でした。MySQLはREPETABLE READの場合実行中にクエリの実行対象が増える事を防ぐためにクエリの実行中はcid=1のレコードのinsertをさせないようにしています。これは例えばlockを取らずにselectで更新対象のprimary keyを取得して、取得したprimary keyで指定して更新すれば「primary keyのcid=1のレコード」だけをロックの対象にする事ができます。ただし、これはselectしてから更新するまでに新規に作られたcid=1のレコードは更新できない事に注意してください。

今回はupdateを例にしましたが、これらの現象はdeleteでも起きます。deleteの場合は特に空振りするとネクストキーロックと言って空振りした前後の空の範囲にもロックを取るため大きな影響がでるリスクがあります。いずれにしてもunique keyで対象をuniqueになるように絞れば問題の影響範囲は最小限に抑えられると思います。

単一プロセスのパフォーマンス向上

パフォーマンスについて考える時は計測するのが大事です。バッチ処理は実行環境を用意するのが難しいケースが多いので大変だと思いますが、さくっと何度でも試せるようにするのが大事です。片道切符のデータ変更のバッチなどの場合は都度データを戻すのが大変だと思うのでテストモードの時はトランザクションをロールバックするようにするとかは手だと思います。こういうのを実装しやすくする意味でもトランザクションの範囲を小さくするのは大切です。

バッチ処理の高速化で重要なのはN+1を防ぐ事です。rails5系の時はactiverecord-importなどが使われていましたがrails6だとinsert_allなど標準のメソッドでバッチ処理が行えます。

言ってる事はread系の画面を作る時の注意と同じ事なのですが、実際に経験しないと「書き込み系の処理だとディスクI/Oの方が問題でしょ？」みたいな感覚になるんじゃないかと思います。ですが実際前職での経験で書き込みのN+1の改善だけでデータ移行のバッチを60倍程度早くできたこともあるので結構大きな効果があると思います。

バッチ処理のN+1で特に難しいのが親子関係のあるようなデータのバッチです。よくやるのが親ごとでループを回して子だけバルクで処理するようなやつです。これだと親がN+1になります。親子ならいいですが孫まで出てくるとナイーブな作りでは高速化できません。

こういうケースの戦略はORMの構造体(ActiveRecordのモデルなど)にインメモリにまずinsert/updateしたいデータを構築して、最後にまとめてバルクで処理するというものです。このやり方なら元のロジックを崩さず、可読性を保った形で全体をバルクで処理する事ができます。

並列実行によるパフォーマンス向上

メンテ中など負荷は気にしなくていいけど短時間で終わらせないといけない時は並列実行できるように設計すると良いです。並列化の戦略は「ユーザー影響の削減」のところと似たような感じです。他のプロセスがユーザーのプロセスかバッチのプロセスかという違いです。

各バッチについて担当範囲がかぶらないように注意して、処理の途中でもギャップロックなどによるロック競合が起きないように気を遣ったロジックにする事が大事です。

負荷対策

N+1を防ぐ、処理の分割、トランザクションを短く、などをきちんと行えば自然と負荷対策になると思っているのでここについては追加で書く事があまりないのですが、定期的にsleepを入れるのはよくやると思います。注意点として初歩的な事ですがsleepのタイミングは必ずトランザクションの外にした方が良いです。トランザクションを手放す前にsleepするとロックがかかったままになるので。

負荷も「試してから考える」でいいと思います。あとは、止めていい処理なら想定外に負荷が上がったら止められるように作るのが大事だと思います。それは冪等に作るとか途中から再実行できるようにするというのと同じような事だと思います。

バッチサイズの制御

負荷のコントロールの難しい内容としてはバッチサイズの制御の話があります。例えば親、子、孫のテーブルのデータを処理して他のテーブルにコピーするような時に親テーブルでバッチサイズを決めてループを回すのがよくやる方法ですが、親のバッチサイズが100としてもバッチ毎に子が1000、孫が10000の事もあれば子も孫も100の事もあります。このようにバッチサイズは親テーブルだけでは十分にコントロールできないケースもあります。こういう時の戦略として前職でやったのは、先に子や孫のidだけ先読みして一定サイズに分割してコピー先のidとのマッピングを永続化(冪等性のところで解説したような手法です)して、親をコピーするフェーズ、子をコピーするフェーズ、孫をコピーするフェーズに分けてそれぞれでバッチサイズを決めてコピーするという事をやりました。外部キーにつっこみたいデータも先に払い出されているのでテーブル毎でのコピーが可能になります。ただ、この作戦はメンテ中などユーザーによってデータが触られないケースでないと難しいと思います。

データ同期で気をつけたいデータ不整合

バッチ処理でよくやるのがあるDBやテーブルのデータに何らかの処理をして別のDBやテーブルにコピーをするというやつです。世間的にニーズが多いのでツールは多い一方で基本的に難しいのでどういうところが難しいかについて書きました。

何が難しいかと言うとソースのデータの変更、削除を追跡するのが難しいからです。 まず単純なappend onlyなログの場合のようにdeleteやupdateが発生しないデータの同期を考えます。この場合データ同期はもし途中でこけてしまってもコピー先の最後のレコードのidやタイムスタンプを見て、それより大きい値のレコードを再同期すれば十分です。

問題はdeleteやupdateが発生した時です。前回のコピーの後に更新された箇所は更新のタイムスタンプがあれば追跡できるかもしれませんが、そのタイムスタンプの更新漏れのリスクはあると思いますし、deleteされた箇所を得るのは困難です。

updateやdeleteの問題を避けるシンプルな戦略としては以下のようなケースが思いつきます。

• 更新を止めて同期する
  • ワンタイムであれば同期中にupdateやdeleteが走らないようにすればこの問題は解決できます。
• ログの形式にする
  • 先ほどappend onlyならコピーが簡単と書いたのですが、updateやdeleteもappend onlyなログの形式に変換するというイメージです。*1
• 毎回全部コピー
  • 全データを毎回全部入れ替えたらdeleteやupdateもちゃんと反映されます。BigQueryみたいなむっちゃ安いストレージに対してならありな戦略だと思います。
• 削除しない
  • updateだけならタイムスタンプを信用して前回の処理以降のタイムスタンプを使えば良いのでまだ実現が可能です。この場合削除は論理削除というものになります。ただ、僕はdeleted_atを用いた論理削除でつらい経験(意図せず削除したはずのデータがユーザーに見えるなど)をした事があり、開発上のデメリットが大きいと感じています。削除済みレコードテーブルを別で用意するとかだとまだ安心できるかもしれないですが僕は経験が無いです。

もしこういった単純な方法が困難な場合はpt-online-schema-changeが行っているようにdatabaseのトリガーを用いる方法やAWS Database Migration Serviceなどがおそらく内部的にやっているバイナリログを用いた方法などで更新を追跡する必要があると思います。

最後に

自分がぱっと思いつく範囲でバッチ処理のテクニックをまとめました。何か参考になることがあれば嬉しいです。

*1:追記: 例えばレコードの更新時にログテーブルに更新内容を一緒に書いて、定期的にそのログテーブルの内容を元に同期する、とかです。ログテーブルに書き込む時点で最初の方に書いた冪等性を崩すようなデータを払い出して保存しておけば冪等にデータを同期できるかなと。