ELK 導入の記録
最近、会社のシステムで集計処理の負荷が大きくなってきました。MySQL クラスタのクエリがボトルネックに達し、インデックス最適化でも集計結果を素早く返せなくなったため、Elasticsearch、Logstash、Kibana というオープンソース三点セットを導入しました。
まず背景を説明しておきます。集計系のクエリと通常の業務クエリは別物です。業務クエリの多くは主キーやインデックスで少数の行を取得するもので、MySQL の得意分野です。一方、集計クエリは大量のデータをスキャンしてグルーピングや集約を行うことが多く、こうした場面では B+ ツリーのインデックスはあまり役に立ちません。データ量が増えてくると、一本の集計 SQL が数秒、時には数十秒もかかり、同じ DB 上の業務リクエストまで巻き込んでしまいます。こうなったときの定番のアプローチが、集計負荷を OLTP データベースから切り離し、集約分析専用のエンジンに任せることです。Elasticsearch の内部は転置インデックスとカラムナ形式の doc values で構成されており、フィルタリングと集約に生来向いています。ELK のこの組み合わせはコミュニティで最も取り組みやすい構成のひとつでもあるので、記録しておく価値があります。
技術選定と役割分担
Logstash については、実は Alibaba の DataX を使うこともできます。機能面では Logstash よりさらに一段上です。私が採用したのは Logstash のままでした。DataX の存在を知ったのが後になってからだったので、あえて置き換えはしませんでした。
両者の位置づけには少し違いがあります。Logstash は Elastic ファミリーのデータパイプラインで、jdbc input プラグインで定期的にデータを取得し、設定ファイルを書けばすぐ動きます。Elasticsearch への出力連携も標準で用意されています。DataX は Alibaba がオープンソース化した異種データソース同期ツールで、プラグインがカバーするデータソースが多く、バッチスループットも強力です。「MySQL から ES への定期増分同期」という単一の経路であれば Logstash で十分であり、乗り換えのための乗り換えをする必要はありません。
三つのコンポーネントの役割分担は明快です。
Logstash が MySQL の集計データを取得・フィルタリングし、Elasticsearch へ増分同期します。
そしてプロジェクトの Java API から Elasticsearch にクエリを投げます。
Kibana はクラスタやインデックスの状況を Web 上で可視化できます。
つまり、Logstash がデータの入口を、Java アプリケーションがデータの出口を担当し、Kibana がクラスタとインデックスの健全性を人の目に見えるようにする——三者は互いに干渉しません。
Elasticsearch のクエリ文自体はかなりシンプルなのですが、公式サイトの例があまり多くない印象で、複雑な集約の多くは自分で模索する必要があります。たとえば多層ネストの aggregation や、集約結果をさらにソートするような書き方は、ドキュメントには最も基本的な例しか載っていないことが多く、実際の業務では Kibana の Dev Tools で少しずつ調整して作り上げることになります。
全体としての学習難易度は高くありません。すぐに使い始められます。
Java 側の接続
Java API の jar には elasticsearch-rest-high-level-client を使いました。
その後、業務シナリオに合わせて自分でファクトリをラップし、開発メンバー向けに数層のコードを抽象化しました。ラップした目的は単純で、各開発者が直接 SearchSourceBuilder を組み立てたり SearchResponse を解析したりするのを避けたかったからです。よく使う条件フィルタ、ページネーション、集約のパターンをいくつかのメソッドに集約し、業務コードはパラメータを渡して結果を受け取ることだけに集中できるようにしました。
API でのクエリと結果セットの取得は操作がやや面倒に感じるかもしれませんが、クエリ文と対照しながら書けば、実はとても理解しやすいものです。high level client のビルダー構造は Query DSL の JSON 構造とほぼ一対一で対応しているので、まず Kibana で DSL を動くまで調整し、それを Java コードに翻訳すれば、ほとんど間違えることはありません。
全体的な印象としてはかなり親切です。どのソフトウェアもインストールすればすぐ使えますが、設定はいくらか変更する必要があります。ここでは詳しく触れませんが、IP・ポート、言語設定、パスワードなどです。
集計に必要なフィールドだけを同期する
会社の主な用途は集計なので、Logstash では重要な集計データだけを取得しています。数百万件のデータから重要な集計フィールドだけを取得すると 100MB にも満たない量になりました。これは ES の集約・集計の速度を大きく向上させます。ですから、テーブル全体を丸ごと取り込むのはおすすめしません。最も重要な集計フィールドだけを取るべきです。
この点は掘り下げる価値があります。Elasticsearch の集約はメモリと doc values 上で実行されるため、インデックスがスリムであるほどセグメントファイルは小さくなり、キャッシュできる割合が高くなって、集約は自然と速くなります。しかも ES はデータベースではないので、「全量の明細を保存する」責務を負う必要はありません。明細は常に MySQL が正であり、ES には集計のディメンションと指標フィールドだけを置きます。壊れてもいつでも再構築できるので、心理的な負担もずっと軽くなります。
全体のフローは図のとおりです。
これは私が会社のメンバー向けに描いたフロー図でもあります。
もし取得するデータ量が非常に大きい場合は、途中に Kafka を挟んでバッファリングと再フィルタリングを行うこともできます。私たちの会社はそこまでのデータ量ではなかったので、他のサーバーリソースを余分に消費することはしませんでした。中間にメッセージキューを一層挟む意義はピークカットと疎結合化にあります。上流の取得と下流の書き込みの速度が一致しないとき、キューが突発的なトラフィックを受け止めてくれますし、途中にクレンジングロジックをもう一段挟むのも容易になります。ただし、コンポーネントが一つ増えるごとに運用コストも一つ増えます。データ量が達していないなら導入すべきではありません。
Kibana は今ではとても完成度が高く、公式の Query DSL でもクエリを書けますし、SQL でも書けます。とはいえ公式は、集約とクエリには Query DSL の使用を推奨しています。
増分同期の実装
増分同期と更新データの同期については、取得対象のデータテーブルに data_version フィールド（楽観ロックの原理によるデータバージョン番号）を設ける方式を採用しました。あるデータが変更されるか新規追加されると、data_version の値が設定されます。私はタイムスタンプを設定しました（会社の業務上の理由からです）。このフィールドには bigint 型を推奨します。timestamp 型では 2028 年までしか使えません。
こうしてデータにバージョンを持たせると、増分・更新データの同期時には data_version が変化したデータだけが取得されます。MySQL サーバー、Logstash、Elasticsearch の負荷を大幅に下げられます。
Logstash は前回の取得時に最後のレコードが持っていた data_version の値を記録できます。これにより次回の取得時には、前回の data_version の値を持って where 条件で有効なデータだけをフィルタリングできます。もちろん SQL の中では data_version でソートする必要があります。Logstash が記録するのは、前回取得時の最後の一件の data_version 値だけだからです。
この裏側にあるのが Logstash jdbc プラグインの tracking column の仕組みです。プラグインは前回実行時の最後のレコードの追跡カラム値をローカルファイルに永続化し、次回の SQL 実行時にパラメータとして where 条件に渡します。だからソートは必須です——結果セットが data_version の昇順で並んでいなければ、記録される「最後の一件」は最大値ではなくなり、次の同期でデータの取りこぼしが発生します。
そうしなければ、毎回の同期のたびに全テーブル同期をするのでしょうか？その負荷は想像に難くありません。そんな方式を採る人はいないでしょう。全テーブル同期は初回の同期時だけで、それ以降はすべて増分・更新データの同期になります。
更新されたデータも同期できるのは、各データが ES に固定のドキュメント ID（通常は MySQL の主キー）で書き込まれるからです。同じデータのバージョンが変わると再度取得され、ES への書き込みは上書き更新になるため、重複ドキュメントは発生しません。
ハマりどころと注意点
1）追跡カラムの型はよく考えましょう。前述のとおり、タイムスタンプをバージョン番号にする場合、フィールドには bigint を推奨します。timestamp 型の値の上限問題を避けるためです。また、タイムスタンプの精度が足りないと、同一秒内の複数回の変更が境界で漏れる可能性があります。業務上それを許容できるか、あるいは自動増分のシーケンスに切り替えるかを検討してください。
2）物理削除は同期できません。data_version 方式で検知できるのは新規追加と変更だけです。MySQL 上で直接 delete された行は結果セットに現れなくなり、ES には古いドキュメントが残ります。業務側で論理削除フラグを使い、削除も一度の「変更」として同期するか、あるいは定期的なインデックス再構築で拾うか、どちらかの対策が必要です。
3）ES を唯一のストレージにしないこと。ES 上の集計インデックスは、いつでも MySQL からリプレイして作り直せる状態にしておくべきです。そうすればマッピング設計を間違えたときやフィールドを追加したいとき、インデックスを削除して Logstash を回し直すだけで済み、複雑なオンライン移行をする必要がありません。
4）増分 SQL のソートと境界条件は必ず検証しましょう。where 条件を「より大きい」にするか「以上」にするか、ソートが効いているか。まず手動で SQL を一度実行し、Logstash の二回の実行で記録された値を突き合わせて、取りこぼしや重複取得がないことを確認するのがおすすめです。
まとめ
今回の ELK 導入の核心的な成果は、集計・集約処理を MySQL クラスタから切り離せたことです。Logstash が data_version に基づいて増分同期を行い、集計に必要なフィールドだけを取得する。Elasticsearch が集約クエリを担い、Java 側は high level client をラップして業務に提供する。Kibana は DSL のデバッグとクラスタ状態の観察を兼ねる。この構成は余計なコンポーネントを持ち込まず、データ量の実際の規模に合わせて取捨選択したものです。必要十分であることが一番です。
COMMENTS