Mafengwo ビッグデータプラットフォームにおける Kafka クラスターの最適化とアプリケーション拡張

Kafka は人気のあるメッセージキューミドルウェアです。大量のデータをリアルタイムで処理でき、高スループット、低レイテンシ、信頼性の高い非同期メッセージ伝送メカニズムという特徴を備えています。異なるシステム間のデータ通信および転送の問題を効果的に解決できます。

Kafka は Mafengwo でも広く使用されており、多くのコアビジネスをサポートしています。この記事では、Mafengwo ビッグデータプラットフォームでの Kafka の適用実践に焦点を当て、関連するビジネスシナリオ、Kafka 適用のさまざまな段階で発生した問題とその解決方法、および将来の計画について説明します。

パート1. アプリケーションシナリオ

ビッグデータプラットフォームにおける Kafka のアプリケーションシナリオの観点から見ると、主に次の 3 つのカテゴリに分けられます。

1 つ目のタイプは、Kafka をデータベースとして使用し、ビッグデータプラットフォーム上でリアルタイムデータのストレージサービスを提供するものです。リアルタイムデータは、ソースと目的という2つの側面から、ビジネス側DBデータ、監視型ログ、埋め込みポイント（H5、WEB、APP、ミニプログラム）に基づくクライアントログ、およびサーバー側ログに分類できます。

2 番目のカテゴリは、データ分析用のデータソースを提供することです。さまざまな追跡ポイントのログは、多次元クエリ、リアルタイム Druid OLAP、ログ詳細など、会社のオフラインデータ、リアルタイムデータウェアハウス、分析システムをサポートおよび接続するためのデータソースとして使用されます。

3 番目のカテゴリは、ビジネス関係者にデータサブスクリプションを提供することです。ビッグデータプラットフォーム内のアプリケーションに加えて、Kafka を使用して、レコメンデーション検索、公共交通機関、ホテル、コンテンツセンターなどのコアビジネス向けに、リアルタイムのユーザー機能計算、リアルタイムのユーザーポートレートトレーニングとリアルタイムのレコメンデーション、不正防止、ビジネス監視とアラームなどのデータサブスクリプションサービスも提供しています。

主な用途を下図に示します。

第2部進化への道

4つのステージ

初期のビッグデータプラットフォームがビジネスログ収集および処理システムとして Kafka を導入した主な理由は、その高スループット、低レイテンシ、複数のサブスクリプション、データバックトラッキングなどの特性により、ビッグデータシナリオのニーズをより適切に満たすことができるためです。しかし、業務量の急激な増加や、登録メカニズムや監視メカニズムの不完全性により問題をすぐに特定できない、一部のオンラインリアルタイムタスクが失敗してもすぐに回復せずメッセージのバックログが発生するなど、業務利用やシステムメンテナンスで発生した問題により、Kafka クラスターの安定性と可用性が課題となり、いくつかの重大な障害が発生しました。

上記の問題を解決することは私たちにとって緊急かつ困難なことです。ビッグデータプラットフォームで Kafka を使用する際のいくつかの問題点に対応するため、クラスターの使用からアプリケーション層の拡張まで、一般的に次の 4 つの段階を含む一連のプラクティスを実行しました。

フェーズ 1: バージョンのアップグレード。プラットフォームデータの生成と消費におけるいくつかのボトルネックと問題に焦点を当て、現在の Kafka バージョンの技術的な選択を行い、最終的にバージョン 1.1.1 を使用することを決定しました。

フェーズ 2: リソースの分離。ビジネスの急速な発展をサポートするために、複数のクラスターの構築とクラスター内のトピック間のリソース分離を改善しました。

第3段階：権限制御と監視アラーム。

まず、セキュリティの面では、初期の Kafka クラスターは裸の実行状態にありました。複数の製品ラインで Kafka を共有しているため、他の事業のトピックを誤って読み取ることでデータセキュリティの問題が発生する可能性が高くなります。そこで、SASL/SCRAM + ACL に基づく認証機能を追加しました。

監視と警告の面では、Kafka は現在、リアルタイムコンピューティングの標準入力データソースとなっているため、ラグバックログとスループットは、リアルタイムタスクが正常であるかどうかを示す重要な指標となっています。そのため、ビッグデータプラットフォームは、Kafka クラスターとユーザーを多次元で監視するための統合 Kafka 監視およびアラームプラットフォームを構築し、「Radar」と名付けました。

フェーズ 4: アプリケーションの拡張。 Kafka を会社のさまざまな事業ラインに公開した当初は、統一された使用仕様がなかったため、一部の事業部門で誤った使用が行われていました。この問題を解決するために、当社はリアルタイムサブスクリプションプラットフォームを構築し、アプリケーションサービスを通じてビジネス関係者に力を与え、データ生成および消費アプリケーション、プラットフォームユーザー認証、ユーザー監視およびアラームなど、多くのリンクのプロセス自動化を実現し、需要側の使用から総合的なリソース管理および制御までの全体的なクローズドループを作成しました。

以下にいくつかの重要なポイントを紹介します。

コアプラクティス

1. バージョンアップグレード

以前、ビッグデータプラットフォームでは、初期の Kafka バージョン 0.8.3 を使用していました。現時点では、Kafka の最新の公式リリースは 2.3 です。したがって、バージョン 0.8 を長期使用する際に発生する多くのボトルネックや問題は、バージョンアップグレードによって解決できます。

たとえば、以前のバージョンで発生した一般的な問題は次のとおりです。

セキュリティのサポート不足: データセキュリティの問題があり、認証と承認によるリソースのきめ細かな管理が不可能
ブローカーのレプリケーション不足: ブローカーがレプリケーション不足の状態にあることが判明しましたが、問題の原因は不明で解決が困難です。
トランザクションメッセージ、べき等メッセージ、メッセージタイムスタンプ、メッセージクエリなどの新しい機能は使用できません。
クライアントのオフセット管理は Zookeeper に依存しており、Zookeeper が過度に使用され、操作と保守の複雑さが増します。
不完全な監視インジケーター: トピック、パーティション、ブローカーデータサイズインジケーターなど、および Kafka Manager などの監視ツールは、低バージョンの Kafka を適切にサポートしていません。

同時に、次のような対象バージョンの機能に関する選定調査を実施しました。

バージョン 0.9 ではクォータとセキュリティが追加されましたが、その中でもセキュリティ認証と承認は私たちが最も関心を持っている機能です。
バージョン 0.10、より細かいタイムスタンプ。オフセットに基づいてデータをすばやく検索し、必要なタイムスタンプを見つけることができます。これは、リアルタイムデータ処理における Kafka データソースに基づくデータ再生にとって非常に重要です。
バージョン 0.11、べき等性とトランザクションのサポート、およびレプリカデータの損失/データの不整合に対するソリューション。
- 冪等性とは、同じパーティションに対してデータが複数回公開された場合、Kafka ブローカーが自動的にデータの重複を排除できることを意味します。
- トランザクションのサポートにより、1 つのトランザクションで複数のメッセージを複数のトピックパーティションにアトミックに公開できるようになります。下流の消費者の多くは、ストリーム処理を実行するために Flink を使用しているため、データ処理と障害回復時には、正確に 1 回のセマンティクスが特に重要です。バージョン 0.11 でのトランザクションのサポートにより、Kafka と対話する Flink アプリケーションがエンドツーエンドで 1 回限りのセマンティクスを実現できるようになります。 EOS のサポートは、トランザクションやリスク管理などのシナリオで重要なサポートとなる、データの信頼性に対する絶対的な要件を満たすことができます。
- リーダーエポック: これにより、レプリカの進行状況を示すために以前は水位に依存していたために発生する可能性のあるデータ損失/不整合の問題が解決されます。
バージョン1.1、保守性が向上しました。たとえば、コントローラーがシャットダウンするときにブローカーをシャットダウンする場合、以前は長くて複雑なプロセスでしたが、バージョン 1.0 では大幅に改善されました。

最終的にバージョン 1.1 が選択された理由は、Camus と Kafka のバージョン間の互換性と、バージョン 1.1 が使用シナリオで重要な新機能をすでにサポートしているという事実に基づいています。ここでは、Linkedin によってオープンソース化されている Camus コンポーネントについて簡単に説明したいと思います。当社のビッグデータプラットフォームでは、主に Kafka データを HDFS にダンプする重要な方法として使用されます。

2. リソースの分離

以前は、ビジネスの複雑さと小規模さのため、ビッグデータプラットフォームの Kafka クラスターの分割は比較的単純でした。その結果、一定期間が経過すると、会社のビジネスデータが混在することになります。特定のビジネストピックを不当に使用すると、一部のブローカーに過負荷が発生し、他の通常のビジネスに影響を及ぼす可能性があります。一部のブローカーに障害が発生した場合でも、クラスター全体に影響が及び、企業全体の業務が利用できなくなるリスクが生じます。

上記の問題に対応するために、クラスター変換の 2 つの側面が実装されました。

独立したクラスターを機能属性ごとに分割する

クラスター内のトピックレベルでのリソース分離

（１）クラスター分割

複数の Kafka 物理クラスターは機能ディメンションに従って分割され、ビジネスを分離し、運用と保守の複雑さを軽減します。

最も重要な追跡データの使用例を挙げると、現在は 3 種類のクラスターに分かれています。各クラスターの機能は次のように定義されます。

ログクラスター: 各端からのデータが収集された後、最初にこのクラスターにデータが配置されます。したがって、Kafka の問題により収集プロセスを中断することはできず、Kafka の可用性に対する要件が高くなります。したがって、クラスターは、消費者がそれを制御できるようにするために外部サブスクリプションを提供しません。同時に、クラスター事業はオフライン収集の源としても機能します。データは Camus コンポーネントを通じて 1 時間単位の粒度で HDFS にダンプされます。この部分のデータは、後続のオフライン計算に使用されます。
フルサブスクリプションクラスター: このクラスターのトピック内のデータのほとんどは、ログクラスターからリアルタイムで同期されます。前述のように、ログクラスター内のデータは一般に公開されていないため、消費とサブスクリプションはクラスター全体で管理されます。現在は主にプラットフォーム内のリアルタイムタスクで使用され、複数の業務ラインのデータを分析し、分析サービスを提供しています。
パーソナライズされたカスタマイズされたクラスター: 前述のように、ビジネスニーズに応じてデータログソースを分割および結合できます。カスタマイズされたトピックもサポートしています。クラスターは、転送後のトピックのランディングストレージのみを提供する必要があります。

クラスターの全体的なアーキテクチャは次のように分類されます。

（２）資源の分離

トピックのトラフィックサイズは、クラスター内のリソース分離の重要な基礎となります。たとえば、当社のビジネスでは、大量の POS ログを持つ 2 つのデータソースは、バックエンドの POS データソースであるサーバーイベントと、エンドの POS データソースであるモバイルイベントです。 2 つのデータを格納するトピックパーティションをクラスター内の同じブローカーのノードに割り当てることは避けたいと考えています。異なるトピックを物理的に分離することで、ブローカー上のトラフィックの偏りを回避できます。

3. 権限制御と監視アラーム

（１）権限制御

冒頭で述べたように、初期の Kafka クラスターはセキュリティ検証が設定されていない裸の状態であったため、ブローカーの接続アドレスがわかっていれば生成と消費を実行でき、深刻なデータセキュリティの問題が発生していました。

一般的に、SASL を使用するユーザーは Kerberos を選択します。ただし、プラットフォーム上の Kafka クラスターの使用シナリオでは、ユーザーシステムは複雑ではなく、Kerberos を使用するのはやややりすぎです。同時に、Kerberos は比較的複雑であり、他の問題を引き起こすリスクがあります。また、暗号化に関しては、すべてイントラネット環境で実行されているため、SSL暗号化は使用されていません。

最終的なプラットフォーム Kafka クラスターは、SASL/SCRAM + ACL の軽量な組み合わせに基づいて SASL を認証方法として使用し、ユーザーの動的な作成を実現してデータのセキュリティを確保します。

（２）監視と警報

これまで、クラスターを使用すると、消費者向けアプリケーションのパフォーマンスが理由もなく低下することがよくありました。問題の原因を分析すると、通常、遅延しているコンシューマーによって読み取られたデータがページキャッシュにヒットしない可能性が高く、ブローカー側のマシンのカーネルが最初にディスクからデータを読み取り、ページキャッシュにロードしてから、結果をコンシューマーに返すことになります。これは、書き込み操作を処理できたはずのディスクが、データの読み取りを必要とすることになり、ユーザーの読み取りと書き込みに影響し、クラスターのパフォーマンスが低下することに相当します。

このとき、遅れている消費者向けアプリケーションを見つけ出し、事前に介入して問題の発生を減らす必要があります。したがって、監視と警告はプラットフォームとユーザーの両方にとって非常に重要です。以下は私たちの実践的なアイデアの紹介です。

全体的な解決策:

全体的なソリューションは、主にオープンソースコンポーネントの Kafka JMX Metrics + OpenFalcon + Grafana に基づいています。

Kafka JMX メトリクス: Kafka ブローカーの内部インジケーターは、JMX メトリクスの形式で外部に公開されます。バージョン1.1.1では、監視ニーズを満たす豊富な監視指標が提供されます。
OpenFalcon: Xiaomi のエンタープライズレベルの高可用性と拡張性に優れたオープンソース監視システム
Grafana: 誰もが使い慣れており、さまざまなメトリックデータソースに接続できるメトリック視覚化システムです。

監視について:

Falcon-agent: 各ブローカーにデプロイされ、Kafka JMXインジケーターレポートデータを解析します。
Grafana: Falcon Kafka Metrics データを視覚化し、クラスター、ブローカー、トピック、コンシューマーの 4 つのロールの監視ダッシュボードを作成するために使用されます。
Eagle: コンシューマーグループのアクティブステータスとコンシューマーグループの Lag バックログを取得します。また、監視・警報システム「Radar」に監視データを提供するための API も提供します。

アラートについて:

レーダーシステム: Falcon と Eagle を通じて Kafka インジケーターを取得し、設定されたしきい値に基づいてアラームを生成する独自開発の監視システム。消費パターンを例にとると、ラグは消費が正常かどうかを測る重要な指標となります。 Lag が増加し続ける場合は対処する必要があります。

問題が発生した場合、コンシューマー管理者だけでなくそのユーザーもそれを知る必要があるため、アラームシステムでもユーザーに通知する必要があります。具体的な方法は、企業向け WeChat アラームロボットを通じて、対応するコンシューマーグループの担当者またはユーザーと Kafka クラスターの管理者に自動的にリマインドすることです。

監視例:

4. アプリケーションの拡張

（１）リアルタイムデータサブスクリプションプラットフォーム

リアルタイムデータサブスクリプションプラットフォームは、Kafka の使用状況の完全なプロセス管理を提供するシステムアプリケーションです。データ生成および消費アプリケーション、プラットフォームユーザーの認証、作業指示の承認によるユーザーの監視とアラームなど、多くのリンクを自動化し、統一された管理と制御を提供します。

中心となる考え方は、Kafka データソースの ID 認証と権限制御に基づいてデータセキュリティを強化し、Kafka ダウンストリームアプリケーションを管理することです。

（２）標準化された申請手続き

生産者や消費者のニーズに関係なく、ユーザーはまず作業注文の形でサブスクリプション申請を提出します。申込情報には、事業内容、テーマ、申込方法等の情報が含まれます。作業指示書は最終的に承認のためにプラットフォームに転送されます。承認されると、ユーザーには承認されたアカウントとブローカーアドレスが割り当てられます。この時点で、ユーザーは通常の生産と消費を実行できます。

（３）監視と警報

プラットフォームの場合、権限はリソースにバインドされ、リソースはプロダクション用の Topics または消費用の GroupTopic になります。権限が割り当てられると、これらのリソースの使用は、リソースのライフサイクル全体を監視するためのレーダー監視システムに自動的に登録されます。

（4）データリプレイ

データの整合性と正確性を考慮すると、Lamda アーキテクチャは現在、ビッグデータでよく使用されるアーキテクチャです。しかし一方で、Lamda アーキテクチャには、リソースの過剰な使用や開発の難易度の高さなどの問題もあります。

リアルタイムサブスクリプションプラットフォームは、消費者グループに任意の場所のリセットを提供し、時間、場所などによるリアルタイムデータの再生をサポートし、Kappa アーキテクチャシナリオをサポートして上記の問題点を解決します。

（５）テーママネジメント

トピック管理を提供する理由は何ですか?簡単な例を見てみましょう。たとえば、ユーザーにクラスター上で独自の Kafka トピックを作成させたい場合、当然、ノード上で直接操作させたくはありません。したがって、前述のサービスの場合、ユーザー向けであれ管理者向けであれ、誰もが SSH 経由でサーバーに接続することは不可能であるため、操作するためのインターフェースが必要です。

そのため、統一されたエントリを作成し、トピックの作成、リソース分離の指定、トピックのメタデータ管理などのトピック管理サービスを導入するための管理機能を提供するサービスが必要です。

（６）データ転用

以前のアーキテクチャでは、ユーザーによる Kafka データ消費の粒度は、各 Kafka トピックが LogSource の完全なデータを格納するというものでした。ただし、使用時には、多くのコンシューマーは各 LogSource のデータの一部のみを消費する必要があります。これは、特定のアプリケーションの複数の埋め込みイベントのデータである可能性があります。下流のアプリケーションが独自のフィルタリングルールを記述する必要がある場合、リソースが無駄になり、使いやすさに問題が生じることは間違いありません。さらに、複数のデータソースを結合して使用する必要があるシナリオもあります。

上記の 2 つの状況に基づいて、ビジネスニーズに応じてトピックの分割、マージ、カスタマイズを実装し、データソース間のデータマージと、アプリコードとイベントコード条件の任意の組み合わせに対するフィルタールールをサポートしています。

パート3. フォローアップ計画

データ重複の問題を解決します。現在のプラットフォームのリアルタイムストリーム処理における障害回復などの要因によって発生するデータ重複の問題を解決するために、Kafka のトランザクションメカニズムと Flink の 2 フェーズコミットプロトコルを組み合わせて、エンドツーエンドで 1 回限りのセマンティクスを実現しようとしています。プラットフォーム上で小規模にテストされており、テストに合格すれば実稼働環境に導入される予定です。

消費者の電流制限。 1 回の書き込みと多数の読み取りのシナリオでは、コンシューマー操作が大量のディスクデータを読み取ると、プロデュースレベルでの他のコンシューマー操作のレイテンシに影響します。したがって、消費フローを制限し、Kafka Quota メカニズムを通じてしきい値の動的な調整をサポートすることも、私たちの将来の方向性です。

シーン拡大。 Kafka 拡張 SDK、HTTP およびその他のメッセージサブスクリプションおよび生成方法に基づいて、さまざまな言語環境とシナリオの使用要件を満たします。

[この記事は51CTOコラムニストMafengwo Technologyによるオリジナル記事です。著者のWeChat公開アカウントはMafengwo Technology（ID：mfwtech）です]

この著者の他の記事を読むにはここをクリックしてください

<<: 金融グレードの分散データベースアーキテクチャの設計を1つの記事で理解する

>>: 知っておくべきパブリッククラウドの最適な帰還ユースケース