Kafka は毎日何千億ものログをどのように処理するのでしょうか?

以前、Kafka の原則について多くの情報を共有しました。今日は、Kafka の数千億のデータ量に基づいた 360 の詳細な実践を見てみましょう。

[[286310]]

画像はPexelsより

この記事は主に以下の内容に焦点を当てています。

• メッセージキューの選択
  
• 360°におけるKafkaの商用化の現状
  
• Kafka クライアント フレームワーク
  
• 高いデータ可用性
  
• 負荷分散
  
• 認証、承認、ACL ソリューション
  
• クォータメカニズム
  
• IDC間のデータ同期
  
• 監視アラーム
  
• オンラインの問題と解決策

メッセージキューの選択

当時は主に以下の次元が考慮されていました。

• コミュニティ活動
• クライアントサポート
• スループット

いくつかのシステムを比較した結果、Kafka が当社の要件を最も満たしていると感じています。今は新しいオープンソースシステム Pulsar があるので、これも試してみようと思います。

Kafka のデザインのハイライトは次のとおりです。

Kafka は高いパフォーマンスとスループットを備えています。 Sendfile と Pagecache を通じてゼロ コピー メカニズムを実装します。シーケンシャルな読み取りおよび書き込み特性により、通常のディスクを使用して高いスループットを実現でき、比較的コスト効率に優れています。

Kafka は、レプリカと ISR メカニズムを通じて高いデータ可用性を保証します。

Kafka クラスターには 2 つの管理ロールがあります。

• コントローラーは主にクラスターの管理を担当します。
• コーディネーターは主にビジネスレベルの管理を行います。

両方の役割は Kafka のブローカーによって実行されるため、フェイルオーバーは非常に簡単です。置き換えるブローカーを選択するだけです。

この観点から見ると、Kafka には分散型設計コンセプトが組み込まれていますが、コントローラー自体もボトルネックになっており、Hadoop の Namenode と比較することができます。

CAP理論については皆さんもよくご存知だと思います。分散システムの実装は CP または AP のいずれかです。

Kafka は実装が比較的柔軟です。さまざまなビジネスでは、独自のビジネス特性に応じて、トピック レベルを CP 指向または AP 指向に構成できます。

事業者間の独立した繰り返し消費をサポートし、再生可能です。

これは Kafka の簡単なアーキテクチャであり、主に次のものに分かれています。

• 生産
• ブローカ
• 消費者側

ログには 3 つのレベルがあります。

• 第1レベルのトピック
• 第 2 レベルのパーティション (各パーティションは並列度です)
• 3番目のレベルはレプリカです（レプリカはパーティションのコピーの数を示します）

360°におけるKafkaの商用化の現状

現在、クラスターには数千億のデータ量と 10,000 ギガバイトのマシンが 100 台以上あります。単一トピックの最大ピークは 600,000 QPS で、クラスターのピークは約 500 万 QPS です。

当社の物理マシンは、24コア/10Gネットワ​​ークカード/128Gメモリ/4T*12 HDDで構成されています。 10G ネットワーク カードと 4T*12 の通常のディスクを使用していることは注目に値します。ディスク スループットとネットワーク スループットが一致します。

さらに、データ量が比較的多いことを考慮すると、SSD ディスクは特に大きくなく、比較的高価です。

ディスク構成構造として JBOD を使用しますが、RAID10 も優れたソリューションです (ディスク コストは 2 倍になります)。

現在の Kafka バージョンは 1.1.1 です。バージョン 0.11 以降を展開することをお勧めします。このバージョンではプロトコルに多くの最適化が施されており、後続の 2.x バージョンと互換性があります。

これは、Kafka のアップストリームおよびダウンストリーム関連コンポーネントです。運用側は主に各種 Kafka クライアント/リアルタイム サービス/Flume/Logstash です。

コンシューマー側は、リアルタイム、オフライン (ETL)、監視の 3 つの部分に分かれています。リアルタイムでは、Spark/Flink/Storm などの主流のフレームワークがあります。オフライン部分については、Flink をベースにした統合ランディング フレームワーク Hamal を開発しました。 Kafka からデータを一度消費するだけで、データを複数のダウンストリーム システム (HDFS、Hbase、Redis など) に配置できるため、繰り返し消費する必要がなくなります。

監視要件もあります。 ES/InfluxDB 関連のログを Kafka に送信し、それらを消費して Grafana で表示していましたが、現在は Prometheus に切り替えました。

Kafka クライアント フレームワーク

なぜこのフレームワークを作成する必要があるのでしょうか?以前は、多くのビジネス部門が、ベア API を使用して Kafka クライアントのロジックを独自に実装していました。

しかし、多くの問題があり、いくつかの異常な状況は完全には捕捉されないでしょう。私たちはこのフレームワークを作成して、すべての詳細を隠して、十分にシンプルなインターフェースを公開しました。

これにより、ビジネス上のエラーの可能性を減らすことができます。ネットワークやクラスタに異常が発生した場合など、極端な状況でも可用性を確保する必要があります。ネットワークまたはクラスターが利用できない場合、データは最初にローカルに保存され、復元時にローカル ディスクから Kafka に復元されます。

私たちは 2 つのフレームワークを実装しました。

• LogProducer は少なくとも 1 回はサポートします。
• LogConsumer は、「少なくとも 1 回」と「正確に 1 回」の 2 つのセマンティクスをサポートします。正確に 1 回、ビジネスでは Rollback インターフェイスを実装する必要があります。

LogProducer フレームワークの基本的な考え方は、メモリ キューを介してログを Kafka に送信することです。 Kafka またはネットワークが利用できない場合は、ローカル ディスクに書き込まれます。同時に、Kafka またはネットワークの可用性をリアルタイムで検出するスレッドも存在します。回復された場合、ディスク ログが読み込まれ、Kafka に送信されます。

メモリを置き換える共有メモリ戦略もサポートしています。共有メモリを使用する目的は、再起動プロセス中に失われるログの数を減らすことです。

LogConsumer のフレームワーク実装では、ブロッキング キューを介してコンシューマー スレッドとワーカー スレッドを分離します。これは、実際には消費ロジックは非常に単純ですが、処理ロジックは非常に複雑であるためです。

このようにして、コンシューマー スレッドとワーカー スレッドに異なる構成を作成でき、ブロッキング キューを通じてバック プレッシャー メカニズムを実装することもできます。

たとえば、ワーカーがリクエストを処理できない場合、ブロッキング キューがいっぱいになり、バック プレッシャーによってコンシューマー スレッドが消費を停止します。

同時に、ワーカー スレッド インターフェイスにインターフェイスを提供し、ユーザーがグローバル オフセット マップに送信できるようにします。

上図に示すように、3 つの複合インターフェースを提供します。ビジネス側のロールバック ロジックがビジネス処理とコミットに実装されている場合、セマンティクスは正確に 1 回となり、デフォルトは少なくとも 1 回セマンティクスとなります。

高いデータ可用性

以前にも述べましたが、Kafka 自体は高いデータ可用性を確保するために Replica+ISR メカニズムを提供していますが、それだけでは不十分である可能性があるため、Rack Aware もサポートする必要があります。

たとえば、Replica=3 の場合、3 つのレプリカが異なる物理ラック上にあることを確認します。この方法では、データが引き続き利用可能である間、同時に最大 2 つの物理ラックで問題が発生することを許容できます。当社の Rack Aware ソリューションは、負荷分散ソリューションとともに実装されます。これについては、後で詳しく説明します。

Kafka は Rack Aware も公式にサポートしており、これは Broker 側で broker.rack パラメータを設定することで実現できることは注目に値します。

ただし、制限があります。各ラックに同じ数のブローカーを割り当てる必要があります。そうしないと、レプリカの分散が偏ってしまいます。実際には、IDC には多数のラックがあり、割り当てられた物理マシンの分布は非常にランダムになる可能性があります。

考えられる解決策としては、仮想ラック グループの概念を採用することです。たとえば、3 つの仮想ラック グループを維持し、適用された物理マシンをこれらの 3 つのグループに追加し、ラック グループ間で割り当てられた物理マシンの数が一貫していることを確認します。

もちろん、異なるラック グループ内の物理マシンは同じ物理ラックを持つべきではありません。

負荷分散

Kafka の負荷分散機能は、Confluent の商用バージョンでのみサポートされています。負荷分散は本質的にレプリカを均等に分散する問題です。

当初、私たちは、以下に示すように、従来の一貫性ハッシュを使用してこの問題を解決したいと考えていました。

その後、従来のワンタイムハッシュではニーズを満たせないことがわかりました。たとえば、ノード node5 を追加する場合、ノード node2 の負荷の一部しか共有できず、グローバル ノード負荷分散は実行できません。

そのため、図に示すように、仮想ノードのワンタイム ハッシュ アルゴリズムに基づくソリューションを実装しました。同じ色は同じ物理マシンに対応し、ハッシュ リング上のすべてのノードは仮想ノードです。

ここには 4 つの物理ノードがあり、そのうち node4 が新しく追加されたノードです。仮想ノードは物理ノードの負荷を十分に均等に分散できるため、ノード 4 をハッシュ リングに追加すると、すべての物理マシンの負荷が共有されます。

アルゴリズムの実装手順は、次の 2 つの主要なステップに分かれています。

① 新しいハッシュ サークルを作成します。vnode_str (hostname-v0 など) を介して MD5 ハッシュを作成し、仮想ノードの vnode_key を取得します。次に、リング ディクショナリを使用して仮想ノードから物理ノードへのマッピングを保存し、vnode_key を sorted_keys リストに追加します。

② ハッシュリングにレプリカを割り当てます。(topic_name+partition_num+replica_num) をキーとして使用し、同じ MD5 ハッシュ アルゴリズムを使用して replica_key を取得します。

次に、sorted_keys 内の replica_key の位置に対してバイナリ検索が実行され、最後に Ring 辞書を使用して物理マシン ノードにマッピングされます。この時点で、レプリカの割り当ては完了です。

このアルゴリズムに基づいて、次の 3 つの問題を解決します。

• 物理ノードを追加するには、少量のデータ移行のみが必要です。
• 異なる構成の物理マシンに重みを設定することで、異機種クラスターの展開をサポートできます。
• Rack Aware for Replica を実装するには、物理​​ノードにラック情報が存在し、物理ノードを Replica に割り当てるときに割り当てられたラック情報が記録されます。

重複がある場合は、vnode_key の位置 + 1 を見つけて次の物理ノードを検索します。 3 つのレプリカの物理ラックが異なることを確認します (Replica=3 の場合)。

リーダーバランス: これは高速かつ低コストの負荷分散方法です。 Kafka はリーダーを通じて読み取りおよび書き込みサービスのみを提供するため、負荷切り替え効果はリーダー切り替えを通じて実現できます。リーダー切り替えのみではデータ同期が不要なので、コストは比較的小さくなります。

ディスクの再バランス: この機能は、Kafka バージョン 1.1.0 以降でサポートされています。 Kafka は、基本的にレプリカ プランを生成し、再割り当てを実行するバランス調整操作用のスクリプトと API をいくつか提供します。

認証、承認、ACL スキーム

新しいクラスターの場合は、実装が簡単なため、SASL ベースの SCRAM ソリューションが推奨されます。

途中で古いクラスタに認証と認可の仕組みを実装したい場合、それは困難であり、各ビジネスに構成の変更を強制する必要があります。同時に、切り替えプロセス中に問題が発生する傾向があります。

以下は、古いクラスターの問題を解決するために実装したホワイトリスト メカニズムの紹介です。まず、古いビジネスをホワイトリストに追加し、新しいビジネスが作業指示プロセスを通じてトピックとコンシューマーのリソース権限を申請してホワイトリストに追加できるようにし、違法な（作業指示のない）トピックとコンシューマーのリソースを定期的に監視します。

同時に、これらのリソースは拒否され、トピックとコンシューマーの読み取りおよび書き込み権限が厳しくなりますが、元のビジネスには影響しません。

クォータメカニズム

クォータは主に、複数のビジネス間のリソース競争の問題を解決するために使用されます。割り当てには 2 つの種類があります。

• 1 つは、ネットワーク帯域幅を制限することです。
• 1 つは、リクエスト レートを制限する (CPU を制限する) ことです。

サービスには、高、中、低の 3 つの優先度を設定しています。高優先度には制限がなく、中優先度は遅延を許容でき、低優先度は極端な場合には停止できます。ツールを使用して、特定の優先度のすべてのサービスを一括制限し、優先度の高いサービスとクラスターのセキュリティを確保できます。

IDC間のデータ同期

まず、IDC 間でデータを同期する必要があるのはなぜでしょうか?同期前は、ビジネスにデータの読み取りと書き込みに関する IDC の概念がない場合があり、そのため IDC 間の読み取りと書き込みが発生しやすく、複数のビジネスで消費と生成が繰り返される可能性があります。

その結果、IDC 間ネットワークの大きな無駄が生じます。また、IDC間ネットワークが安定せず、異常が発生することもあり、業務がうまく処理できないこともあります。

上記の課題を解決するために、IDC間データ同期サービスを統一しました。まず、ビジネスは IDC 内でのみ読み取りと書き込みが可能であり、IDC 間の読み取りと書き込みは許可されないことに同意します。 IDC 間のデータが必要な場合は、弊社に申請し、Mirrormaker を通じてコピーを同期する必要があります。

これには 2 つの利点があります。

• まず、異常がビジネスに与える影響を防ぎます。
• 2 番目に、IDC 間の帯域幅を節約します (同期メカニズムにより、データのコピーが 1 つだけ送信されるようにすることができます)。

また、このサービスを Marathon/Mesos に基づいてパスベースにし、サービス SLA を改善しました。

監視アラーム

当社の監視および警告プラットフォームは次のとおりです。

• Jmxエクスポーター+Promehteus+Grafanaをベースにチャート表示を行います。 Jmx エクスポーターは各ブローカーにデプロイされ、Prometheus がデータをプルし、最終的に Grafana を通じて表示します。
• Kafka Manager に基づいて一時的な指標を監視します。
• Burrow に基づいて消費者の遅延を監視します。
• アラームは、Zabbix と同様に 360 によって内部的に実装されたコンポーネントである Wonder に基づいています。

オンラインの問題と解決策

ディスク障害: Smartctl を使用して監視します。まず、ステータスが合格である必要があります。次に、197 Current_Pending_Sector の属性値が 100 より大きくできないことを確認します。100 より大きい場合、ディスクの読み取りおよび書き込みパフォーマンスに問題がある可能性があります。

bootstrap.servers パフォーマンスのボトルネック: このパラメータは、Kafka クライアントにルックアップ サービスを提供するプロキシとして機能する複数のブローカーを構成できます。

クラスターが大きく、クライアントの数が多い場合、これらのプロキシ ロールのブローカーにかかる負荷は非常に大きくなります。この問題を解決するために、bootstrap.servers パラメータに VIP 構成を作成しました。

各 VIP は任意の数のブローカーにバインドできるため、クライアントが構成を変更することなくプロキシを動的に拡張または縮小できます。

消費者は再起動するが消費しない: ビジネス フィードバックにより、消費が停止し、再起動しても問題は解決されないことが示されています。その後、0.11 より前のバージョンではバグであることが判明しました。

 https://issues.apache.org/jira/browse/KAFKA-5413

その理由は、ログ クリーナー スレッドがハングし、圧縮が停止してしまうためです。 __consumer_offsets トピックのボリュームが非常に大きく、ブローカーのリロード時間が特に長くなります。この期間中はサービスは停止されます。

解決策は2つあります。

• まず、Kafka 0.11以降のバージョンにアップグレードします
• 2 番目の解決策は、オフセットを新しいコンシューマー グループに移行することです (問題のあるコーディネーターを回避します)。

ヤン・スーペン

[[286313]]

Qihoo 360 のビッグデータ アーキテクチャ運用および保守の専門家である Yan Suopeng 氏は、インフラストラクチャとビッグデータ開発の分野で 10 年の経験を持っています。 2013年に360商用化チームに入社し、メッセージミドルウェアの開発・運用を担当。ビッグデータアーキテクチャ、マイクロサービスアーキテクチャ、リアルタイムコンピューティングプラットフォーム、機械学習プラットフォーム、監視システムなどのインフラストラクチャの構築にも携わっています。彼は、商業化チームに安定した効率的な基本サービスを提供することに尽力しています。