株式会社kubell 技術基盤開発部の佐藤（@Satoooooooooooo）です。
こちらは kubell Advent Calendar 2024 シリーズ3 12月25日分の記事です。
この記事ではプロダクトオーナー目線でふりかえる大規模DBリプレイスの続編として、どのようにデータ移行をしたのか書いてみようと思います。
※プロジェクトの概要については前回の記事を参照してください creators-note.chatwork.com
メッセージDBとは？
メッセージDBのデータ移行について説明する前に、メッセージDBについて簡単に説明します。
Chatwork のサービスにはメッセージデータの永続化を責務とするFalcon というサブシステムがあり、外部のサブシステムに対して Web API 形式のインタフェース(message-write-apiとmessage-read-api)を公開しています。
Falcon は内部的に CQRS+ES を採用していて、クエリ側のデータストアのことを「メッセージDB」と呼んでいます。
コマンド側からはメッセージの作成・編集・削除イベントを Kafka に書き込み、そのイベントストリームを read-model-updater が吸い上げてメッセージDBを更新する仕組みになっています。
移行方法
HBaseからDynamoDBへのデータ移行方法を検討するにあたって、主に次の2点を考慮しました。
- システム停止時間
- 新DB切り替え時のリスク管理
メッセージDBの移行を検討する上で一番ナイーブなやり方はシステムを停止してその時点のデータをリプレイス先に投入する方法になります。 しかし、メッセージDBには、kubellが「Chatwork」の提供を開始して以来蓄積してきた約130億件のメッセージが格納されているため、これをDynamoDBに移行するだけでも数日間かかります。 数日間のシステム停止というのは世の中のシステムの大半が受け入れ難い条件ですし、ビジネスチャットアプリの特性上、移行に伴うシステム停止はできるだけ短くする必要がありました。
またパフォーマンス検証は時間をかけて実施しましたが、本番環境のトラフィックでも想定通りのパフォーマンスを達成できるか不安が残っていました。 そのため、段階的にリプレイス先のDynamoDBに負荷をかけていき、もし何か問題が起こればHBaseに戻せるようにする必要がありました。
これらのことを考慮して、今回のプロジェクトでは以下のような移行ステップを踏むことにしました。
Step.1 初期移行
移行開始時点のメッセージDB(旧)スナップショットからHBaseに書き込まれたデータを読み込み、DynamoDBに移行します。
HBaseのデータをDynamoDBに移行する方法はいくつかありますが、以下の理由で自作のツールを採用しました。
- 別件対応で既に途中まで実装されていた移行ツールがあった
- 途中で失敗した場合に再開可能
- 進捗状況がわかる
自作のツールは以下のような仕組みでルーム単位でメッセージを移行します。
このツールはAmazon EKS上で動作し、Pod数を増減させることでスループットを調整できるように設計しています。
Step.2 差分移行
初期移行開始後にKafkaに追加されたイベントをDynamoDBに連携します。初期移行開始直後のイベントから最新のイベントまで同期できれば差分移行完了です。
Step.3 message-read-apiの段階的なDynamoDBへの移行
messase-read-apiの読み込み先を徐々にDynamoDBに移していきます。最初はチーム内から始めて、社内のチャット、社外の20%、50%と徐々にDynamoDBへのリクエスト比率を増やすようにしました。
そして100%DynamoDBにリクエストするようにした後も、何かあった時に再度HBaseに戻せるようにしばらくHBaseを稼働させるようにしました。
Step.4 移行完了
一ヶ月並行稼働させて問題がなければHBaseを削除します。
個人的に印象に残ったこと
初期移行+差分移行という方式はkubellでは実績のある方法でしたが、今回DynamoDBに移行するにあたって初期移行を何日で終わらせることができるかが一つの焦点でした。 FalconではKafkaのイベント保持期限は7日間となっており、この保持期限を過ぎると先頭からイベントが消失し、差分移行が不完全になります。 件数確認や作業バッファを考慮すると大体4日以内には初期移行を完了する必要がありました。
メッセージ件数130億件、1つのメッセージを書き込むのに必要な書き込みユニットの平均をxと置くと、初期移行を4日間で完了するために必要なスループットはざっくり 37615x unit/s 程度になります。 メッセージの多くはDynamoDBの書き込み単位である1KB以内に収まるので、xは大きくても2以下で収まりそうです。このことから初期移行では消費WCU 80000 (80000 unit/s) 程度のスループットを狙うことにしました。
※WCU(書き込みオペレーションのキャパシティユニット)については以下の記事を参照してください。 docs.aws.amazon.com
さて実際にやってみると、消費WCU 80000を達成するのは簡単ではありませんでした。 Pod数を20、30、40...と増やすに従って、メトリクス上でスロットリングが増加し、全体のスループットが伸び悩むという事象が観測されました。 DynamoDBでは1つのパーティションに割り当てられるWCUを超える書き込み要求があると、スロットリングが発生し、超過分の書き込み要求は失敗します。
この問題に対して、当初はスロットリングの発生が全体のスループットを低下させる原因と考えて、1回のバッチ書き込みで複数のパーティションキーに分散して書き込むなどしてスロットリングを回避することを検討しました。
しかし、移行後の件数確認で相違が発生した時などにデバッグをすることを考えると、初期移行ツールはできるだけシンプルであることが望ましいため、別の方法も検討する必要がありました。
そこで、初期移行ツールが目指す状態(=書き込みで最大の性能が出ている状態)を改めて整理してみることにしました。
DynamoDBでは1つのパーティションに割り当てられるWCUは最大でも 1000 なので、もし移行中に特定のパーティションに書き込みが集中するような(極端な例だと全てのPodが1つのパーティションに書き込む)状況では、書き込みが行われないパーティションの影響により全体のスループットが大きく落ち込むことになります。
一方で、全てのパーティションでWCUの限界まで書き込み要求があれば、全体として最大のスループットが出せていそうです。
では、全てのパーティションでスロットリングが発生している状態はどうでしょうか？
パーティションに割り当てられたWCUを使い切るまでは書き込み要求は成功するはずですから、これも全体として最大のスループットが出せそうです。
どうやら実際のアプリケーションではスロットリングの発生は無視できない問題であるため、無意識のうちにスロットリングを回避すべきものだと考えてしまっていたようです。 しかしDynamoDBへのデータ移行においてはスロットリングの有無は本質ではなく、単純に全てのパーティションのWCUを限界まで消費できているかどうかが問題でした。
ここまで理解すれば、あとはスロットリングの増加に臆することなく並列数(Pod数)を増やすだけです。 Pod数を十分に増やせば、全てのパーティションが常に複数のPodから書き込みが行われる確率が高くなるので、WCUを限界まで消費できるはずです。 実際にPod数を150まで増やしてみると、プログラムを改修することなく目標に近いスループットを達成することができました。パワー！！
後から振り返ると何故悩んでいたのかわからない簡単な話ではありますが、問題を整理することでシンプルな方法で問題を解決できたことは大きな学びになりました。
終わりに
今回のメッセージDBのデータ移行はFalconのCQRS+ESの仕組みを最大限活用したものです。 もしFalconがこのような仕組みになっていなかったら、ここまで短期間で目的を達成することはできませんでした。 このことを考えてもFalconは優れたアーキテクチャだと感じました。先人に感謝です。
私たちが技術的に良い判断をできたかはわかりませんが、この記事が同じような状況で困っている誰かの役に立てたら嬉しいです。
