この記事では、Kafkaのワークフローとストレージの仕組みについて説明します。

1. パーティションが指定されている場合、指定された値がパーティション値として直接使用されます。
2. パーティション値が指定されていないがキーが利用可能な場合、パーティション値はキーのハッシュ値の係数とトピックのパーティション数を取得することによって取得されます。
3. パーティション値もキー値もない場合は、最初の呼び出しでランダムな整数が生成され (その後の呼び出しごとに整数が増加します)、この値の係数とトピックで使用可能なパーティションの合計数を取得することでパーティション値が取得されます。これは、ラウンドロビン アルゴリズムと呼ばれることがよくあります。

2.3 レプリケーション

同じパーティションに複数のレプリケーションが存在する場合があります (server.properties 構成の default.replication.factor=N に対応)。レプリケーションがないと、ブローカーがクラッシュすると、そのパーティション内のすべてのデータを消費できなくなり、プロデューサーはパーティションにデータを保存できなくなります。レプリケーションの導入後、同じパーティションに複数のレプリケーションが存在する場合があり、その際にこれらのレプリケーションの中からリーダーを選択する必要があります。プロデューサーとコンシューマーはこのリーダーとのみ対話し、他のレプリケーションはリーダーからデータをコピーするフォロワーとして機能します。

2.4 データ信頼性保証

プロデューサーによって送信されたデータが指定されたトピックに確実に送信されるようにするには、トピックの各パーティションがプロデューサー データを受信した後、プロデューサーに ack (確認応答) を送信する必要があります。プロデューサーが ACK を受信した場合は次のラウンドを送信し、受信しなかった場合はデータを再送信します。

a) レプリカ データ同期戦略には主に 2 つあります。

Kafka が 2 番目のソリューションを選択したのは、次の理由によるものです。

• n 個のノードの障害を許容するために、最初のソリューションでは比較的多数のレプリカが必要になります。 Kafka の各パーティションには大量のデータがあるため、最初のソリューションでは大量のデータ冗長性が発生します。
• 2 番目のソリューションのネットワーク レイテンシは比較的高いですが、ネットワーク レイテンシが Kafka に与える影響は比較的小さくなります。

b) 情報通信技術

2 番目のソリューションを採用した後、次のシナリオを想像してください。リーダーがデータを受信し、すべてのフォロワーがデータの同期を開始しますが、1 つのフォロワーがハングアップし、リーダーと同期できなくなります。その後、リーダーは ack を送信する前に同期が完了するまで待機する必要があります。この問題を解決するにはどうすればいいでしょうか?

リーダーは、動的な同期レプリカ セット (ISR) を維持します。これは、リーダーと同期されているフォロワーのセットを意味します。 ISR 内のフォロワーがデータ同期を完了すると、リーダーはフォロワーに ack を送信します。フォロワーが長時間リーダーとデータを同期できない場合、フォロワーは ISR から追い出されます。時間のしきい値は、replica.lag.time.max.ms パラメータによって設定されます。リーダーが失敗すると、ISR から新しいリーダーが選出されます。 (以前は別のパラメータがありましたが、バージョン 0.9 以降では replica.lag.max.messages パラメータは削除されました)

c) ack応答メカニズム

重要度の低いデータの場合、データの信頼性要件はそれほど高くなく、少量のデータ損失は許容されるため、ISR 内のすべてのフォロワーがデータを正常に受信するまで待つ必要はありません。

したがって、Kafka はユーザーに 3 つの信頼性レベルを提供します。ユーザーは、信頼性とレイテンシ要件のトレードオフに基づいて、次の acks パラメータ構成を選択できます。

0: プロデューサーはブローカーの ack を待機しません。この操作により、レイテンシが最小になります。ブローカーは、データを受信するとすぐにディスクに書き込む前に戻ります。ブローカーに障害が発生すると、データが失われる可能性があります。

1: プロデューサーはブローカーの ack を待ちます。パーティション リーダーは、データをディスクに正常に保存した後、ack を返します。フォロワーが正常に同期される前にリーダーに障害が発生すると、データが失われます。

-1 (すべて): プロデューサーはブローカーの ack を待機し、パーティションのリーダーとフォロワーがすべてデータをディスクに正常にアップロードした後にのみ ack を返します。ただし、フォロワーの同期が完了してからブローカーが ack を送信する前にリーダーに障害が発生すると、データの重複が発生します。

d) トラブルシューティング

Kafka クラスター内のフォロワーの長さが常にリーダーの長さと一致することを保証できないため (つまり、データ同期には遅延があります)、リーダーに障害が発生し、フォロワーが新しいリーダーとして選出されると (障害が発生した元のリーダーが回復してフォロワーになります)、リーダーのデータがフォロワーよりも少なくなる可能性があります。一貫性のないデータ量によって生じる混乱を解決するために、Kafka は次の概念を提案しています。

• LEO (ログ終了オフセット): 各レプリカの最後のオフセットを指します。
• HW (High Wather): コンシューマーが確認できる最大オフセットと、ISR キュー内の最小の LEO を指します。

コンシューマーがリーダーと通信する場合、HW より前のデータのみを消費でき、HW より後のデータはコンシューマーには表示されません。

このルールの場合:

• フォロワーが失敗した場合: フォロワーが失敗すると、一時的に ISR から追い出されます。フォロワーが回復すると、フォロワーはローカル ディスクに記録された最後の HW を読み取り、ログ ファイルのうち HW よりも高い部分を切り取り、HW からリーダーとの同期を開始します。フォロワーの LEO がパーティションの HW 以上の場合、つまりフォロワーがリーダーに追いつくと、ISR に再参加できます。
• リーダーが失敗した場合: リーダーが失敗した後、ISR から新しいリーダーが選択されます。その後、複数のレプリカ間でデータの一貫性を確保するために、残りのフォロワーはまず HW より高いログ ファイルの部分を切り取り、次に新しいリーダーからのデータを同期します。

したがって、データの一貫性は、データが失われたり重複したりしないことを保証するものではなく、ack によって制御されます。 HW ルールはレプリカ間のデータの一貫性のみを保証できます。

2.5 正確に1回だけのセマンティクス

サーバーの ACK レベルを -1 に設定すると、プロデューサーとサーバーの間でデータが失われないことが保証されます (つまり、「少なくとも 1 回」のセマンティクス)。対照的に、サーバーの ACK レベルを 0 に設定すると、プロデューサーからの各メッセージが 1 回だけ送信されるようになります (つまり、「At Most Once」セマンティクス)。

「少なくとも 1 回」では、データが失われないことは保証されますが、データが重複しないことは保証されません。対照的に、「At Most Once」ではデータが重複しないことを保証できますが、データが失われないことは保証できません。ただし、トランザクション データなどの非常に重要な情報については、下流のデータ コンシューマーは、データが重複したり失われたりしないこと、つまり Exactly Once セマンティクスを要求します。 0.11 より前のバージョンの Kafka では、これを実行する能力がなく、データが失われていないことを確認してから、下流のコンシューマーでデータのグローバル重複排除を実行することしかできませんでした。ダウンストリーム アプリケーションが複数ある場合は、それぞれをグローバルに重複排除する必要があり、パフォーマンスに大きな影響を与えます。

Kafka バージョン 0.11 では、べき等性という主要な機能が導入されました。いわゆるべき等性とは、プロデューサーがサーバーに重複したデータを何回送信しても、同じ結果が返されることを意味します。サーバーは 1 つのレコードのみを保持します。冪等性と「少なくとも1回」の意味論を組み合わせると、Kafkaの「正確に1回」の意味論が構成されます。つまり、「少なくとも1回」+「冪等性」=「正確に1回」となります。

べき等性を有効にするには、プロデューサー パラメータで enable.idompotence を true に設定するだけです。 Kafka のべき等性の実装では、実際には、もともと下流で実行する必要があった重複排除がデータの上流に移動します。冪等性が有効なプロデューサーには初期化中に PID が割り当てられ、同じパーティションに送信されるメッセージにはシーケンス番号が付加されます。ブローカー側は

ただし、再起動すると PID が変更され、パーティションごとに主キーが異なるため、パーティション間のセッションでは冪等性によって Exactly Once を保証することはできません。

3. ブローカーがメッセージを保存する

3.1 保管方法

トピックを 1 つ以上のパーティションに物理的に分割し (server.properties の num.partitions=3 構成に対応)、各パーティションは物理的にフォルダーに対応します (フォルダーにはパーティションのすべてのメッセージとインデックス ファイルが格納されます)。

3.2 ストレージ戦略

Kafka は、メッセージが消費されたかどうかに関係なく、すべてのメッセージを保持します。古いデータを削除するには、次の 2 つの戦略があります。

時間ベース: log.retention.hours=168

サイズに基づく: log.retention.bytes=1073741824 Kafka が特定のメッセージを読み取る時間の複雑さは O(1) であり、ファイル サイズとは無関係であるため、ここで期限切れのファイルを削除しても Kafka のパフォーマンスの向上には関係ないことに注意してください。

4. Kafka の消費プロセス

Kafka コンシューマーはプル モードを使用してブローカーからデータを消費します。対照的に、プッシュ モードでは、メッセージの送信レートがブローカーによって決定されるため、消費レートが異なるコンシューマーに適応することが困難です。メッセージをできるだけ早く配信することが目標ですが、これにより、消費者がメッセージを処理する時間が足りなくなる可能性が高くなります。プル モードでは、コンシューマーの消費容量に基づいて適切な速度でメッセージを消費できます。

プル モードの欠点は、Kafka にデータがない場合、コンシューマーがループに陥り、空のデータを返し続ける可能性があることです。これを回避するために、プル リクエストには、コンシューマー リクエストがデータの到着を待機する「ロング ポーリング」でブロックできるようにするパラメーターがあります (オプションで、転送サイズが大きいことを確認するために指定されたバイト数まで待機するか、待機タイムアウトを渡すこともできます)。

4.1 消費者グループ

コンシューマーはコンシューマー グループの形式で動作し、1 人以上のコンシューマーがグループを形成して一緒にトピックを消費します。各パーティションは、グループ内の 1 つのコンシューマーのみが一度に読み取ることができますが、複数のグループが同時にパーティションを消費することは可能です。図では、3 人のコンシューマーで構成されるグループがあり、1 人のコンシューマーがトピック内の 2 つのパーティションを読み取り、他の 2 人のコンシューマーがそれぞれ 1 つのパーティションを読み取ります。コンシューマーはパーティションを読み取ります。コンシューマーはパーティションの所有者とも呼ばれます。

この場合、コンシューマーは水平方向にスケーラブルな方法で大量のメッセージを同時に読み取ることができます。さらに、コンシューマーに障害が発生した場合、他のグループ メンバーは、以前に障害が発生したコンシューマーによって読み取られたパーティションを自動的に読み取って負荷分散します。

コンシューマ グループの最も重要な機能は、ブロードキャスト機能とユニキャスト機能を実現することです。コンシューマー グループは、サブスクライブするトピックの各パーティションが、そのコンシューマー グループに属する単一のコンシューマーによってのみ消費されることを保証できます。異なるコンシューマー グループが同じトピックをサブスクライブする場合、これらのコンシューマー グループは互いに独立しており、互いに干渉しません。

メッセージを複数のコンシューマーが消費するようにしたい場合は、これらのコンシューマーを異なるコンシューマー グループに配置することができます。これは実際にはブロードキャストの効果です。メッセージを 1 つのコンシューマーのみで消費したい場合は、これらのコンシューマーを同じコンシューマー グループに配置することができます。これは、実際にはユニキャストの効果です。

4.2 パーティション割り当て戦略

コンシューマー グループには複数のコンシューマーが存在し、トピックには複数のパーティションがあるため、どのパーティションがどのコンシューマーによって消費されるかを決定するというパーティション割り当ての問題が必然的に発生します。

Kafka には、RoundRobin と Range の 2 つの割り当て戦略があります。

ラウンドロビン

RoundRobin はポーリングを意味します。たとえば、ConsumerA、ConsumerB、ConsumerC の 3 つのコンシューマーから構成されるコンシューマー グループがあり、これらのコンシューマーは同時に TopicA のメッセージを消費します。 TopicA は 7 つのパーティションに分かれています。 RoundRobin 割り当て戦略を採用する場合、プロセスは次のようになります。

画像: mrbird.cc

この世論調査の方法は簡単に理解できるはずです。しかし、コンシューマー グループが複数のトピックの複数のパーティションを消費するとどうなるでしょうか?たとえば、ConsumerA と ConsumerB という 2 つのコンシューマーで構成されるコンシューマー グループがあり、これらのコンシューマーは TopicA と TopicB からのメッセージを同時に消費します。 RoundRobin 割り当て戦略を採用する場合、プロセスは次のようになります。

注: TAP0 は TopicA Partition0 のパーティション データを示します。

この場合、割り当てにはラウンドロビン アルゴリズムが使用され、複数のトピックはそれぞれのパーティションを含む全体として扱われます。たとえば、Kafka-clients 依存関係では、対応するオブジェクトは TopicPartition です。次に、これらの TopicPartitions はハッシュ値に従ってソートされ、ラウンドロビン方式でコンシューマーに割り当てられます。

しかし、これによって問題が発生します。上図の消費者グループで、ConsumerA が TopicA のみをサブスクライブし、ConsumerB が TopicB のみをサブスクライブしている場合、RoundRobin ポーリング アルゴリズムを使用した後、ConsumerA が TopicB パーティションでメッセージを消費し、ConsumerB が TopicA パーティションでメッセージを消費する可能性があります。

要約すると、ラウンドロビン アルゴリズムは、コンシューマー グループ内のコンシューマーによってサブスクライブされるトピックが同じである場合にのみ適用されます。同時に、ラウンドロビン アルゴリズムを使用すると、コンシューマー グループ内のコンシューマーによって消費されるメッセージの数は最大 1 だけ異なることがわかります。

範囲

Kafka はデフォルトで範囲割り当て戦略を使用します。名前が示すように、Range は範囲による分割を意味します。

たとえば、ConsumerA、ConsumerB、ConsumerC の 3 つのコンシューマーから構成されるコンシューマー グループがあり、これらのコンシューマーは同時に TopicA のメッセージを消費します。 TopicA は 7 つのパーティションに分かれています。範囲割り当て戦略を採用する場合、プロセスは次のようになります。

TopicA と TopicB からのメッセージを同時に消費する、ConsumerA と ConsumerB という 2 つのコンシューマーで構成されるコンシューマー グループがあるとします。範囲割り当て戦略を採用する場合、プロセスは次のようになります。

範囲アルゴリズムは、複数のトピック パーティションを全体として扱いません。

上記の例から、Range アルゴリズムの欠点をまとめることができます。つまり、同じコンシューマー グループ内のコンシューマーによって消費されるメッセージの数は大きく異なる可能性があるということです。

4.3 オフセットメンテナンス

消費者は消費プロセス中に停電やその他の障害を経験する可能性があるため、回復後は障害前の位置から消費を継続する必要があります。したがって、コンシューマーは、障害が回復した後も消費を継続できるように、消費したオフセットをリアルタイムで記録する必要があります。

Kafka バージョン 0.9 より前では、コンシューマーはデフォルトでオフセットを Zookeeper に保存していました。バージョン 0.9 以降では、コンシューマーはデフォルトで Kafka の組み込みトピック (_consumer_offsets) にオフセットを保存します。

 > bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic starfish --from-beginning  
 1つ

トピックを消費した後、kafka-logs ディレクトリを確認すると、さらに 50 個のパーティションが見つかります。

デフォルトでは、__consumer_offsets には 50 個のパーティションがあります。システム内に多数のコンシューマー グループがある場合、このコマンドの出力は非常に大きくなります。

5. Kafka トランザクション

Kafka はバージョン 0.11 からトランザクション サポートを導入しました。トランザクションにより、Kafka は Exactly Once セマンティクスに基づいて、パーティションやセッションを越えて生成と消費が可能になり、すべてが成功するか、すべてが失敗することを保証できます。

5.1 プロデューサー取引

パーティション間およびセッション間のトランザクションを実装するには、グローバルに一意の TransactionID を導入し、プロデューサーによって取得された PID を Transaction ID にバインドする必要があります。この方法では、プロデューサーを再起動すると、進行中の TransactionID を通じて元の PID を取得できます。

トランザクションを管理するために、Kafka は新しいコンポーネントである Transaction Coordinator を導入しました。プロデューサーは、トランザクション コーディネーターと対話して、トランザクション ID に対応するタスク ステータスを取得します。トランザクション コーディネーターは、すべてのトランザクションを Kafka の内部トピックに書き込む役割も担っています。これにより、サービス全体が再起動された場合でも、トランザクションの状態が保存され、進行中のトランザクションの状態を復元して続行できます。

5.2 消費者取引

Consumer の場合、トランザクションの保証は比較的弱く、特に Commit メッセージが正確に消費されるという保証はありません。これは、コンシューマーがオフセットを通じてあらゆる情報にアクセスでき、異なるセグメント ファイルは異なるライフ サイクルを持つため、同じトランザクション内のメッセージが再起動後に削除される可能性があるためです。

参照:

シリコンバレーの Kafka チュートリアル

いくつかの写真は mrbird.cc から引用しました

https://gitbook.cn/books/5ae1e77197c22f130e67ec4e/index.html