Kafka のレプリケーションメカニズムをご存知ですか?

[[379096]]

日常的な開発プロセスでは、フロー制限とピークシェービングを実装するために Kafka が使用されます。ただし、Kafka はデータの損失を防ぐために複数のコピーを保存することがよくあります。その仕組みが何なのかご存知ですか？この記事ではそれについて説明します。

1. Kafka クラスター

Kafka は Zookeeper を使用してクラスター メンバー (ブローカー) に関する情報を維持します。各ブローカーには一意の識別子 broker.id があり、クラスター内でブローカー自身を識別するために使用されます。これは、構成ファイル server.properties で構成することも、プログラムによって自動的に生成することもできます。以下は、Kafka ブローカーのクラスターを自動的に作成するプロセスです。

• 各ブローカーが起動すると、Zookeeper の /brokers/ids パスの下に一時ノードが作成され、そのノードに broker.id が書き込まれ、ブローカー自体がクラスターに登録されます。
• ブローカーが複数ある場合、すべてのブローカーが Zookeeper 上に /controller ノードを作成するために競合します。 Zookeeper 上のノードは重複しないため、ブローカーは 1 つだけ正常に作成されます。このとき、ブローカーはコントローラー ブローカーと呼ばれます。他のブローカーの機能に加えて、トピック パーティションとそのレプリカのステータスの管理も担当します。
• ブローカーがクラッシュしたり自発的に終了したりして、ブローカーが保持している Zookeeper セッションがタイムアウトすると、Zookeeper に登録されたウォッチャー イベントがトリガーされ、Kafka は対応するフォールト トレランス処理を実行します。コントローラー ブローカーがクラッシュした場合も、新しいコントローラーの選択がトリガーされます。

2. コピーメカニズム

高可用性を確保するために、Kafka パーティションが複製されます。 1 つのレプリカが失われた場合でも、パーティション データは他のレプリカから取得できます。ただし、これには、対応するレプリカのデータが完全である必要があり、これが Kafka データの一貫性の基礎となるため、特別な管理のためにコントローラー ブローカーが必要になります。以下では、Kafka のレプリケーション メカニズムについて詳しく説明します。

2.1 パーティションとレプリカ

Kafka トピックは、Kafka の最も基本的なストレージ ユニットである複数のパーティションに分割されます。各パーティションには複数のレプリカを含めることができます (トピックの作成時に replication-factor パラメータを使用して指定できます)。レプリカの 1 つはリーダー レプリカであり、すべてのイベントはリーダー レプリカに直接送信されます。他のレプリカはフォロワー レプリカであり、リーダー レプリカとのデータの一貫性を保つために複製する必要があります。リーダー レプリカが使用できない場合は、フォロワー レプリカの 1 つが新しいリーダーになります。

2.2 ISRメカニズム

各パーティションには、同期された利用可能なレプリカをすべて保持する ISR (同期レプリカ) リストがあります。リーダー レプリカは同期レプリカである必要があり、フォロワー レプリカの場合は、同期レプリカと見なされるために次の条件を満たす必要があります。

• Zookeeper とのアクティブなセッションがあるため、ハートビートを Zookeeper に定期的に送信する必要があります。
• 指定された時間内に、低遅延でリーダー レプリカからメッセージを受信します。

レプリカが上記の条件を満たさない場合、ISR リストから削除され、条件が満たされるまで再度追加されることはありません。

トピック作成の例を次に示します。--replication-factor を使用して、レプリケーション係数 3 を指定します。作成が成功したら、--describe コマンドを使用して、パーティション 0 にレプリカ 0、1、2 の 3 つのレプリカがあり、3 つのレプリカすべてが ISR リストにあり、1 がリーダー レプリカであることを確認します。

2.3 不完全なリーダー選出

レプリケーション メカニズムの場合、ブローカー レベルでオプションの構成パラメーター unclean.leader.election.enable があります。デフォルト値は fasle であり、不完全なリーダー選出が禁止されることを意味します。これは、リーダー レプリカに障害が発生し、ISR 内に利用可能な他のレプリカがない場合に、不完全に同期されたレプリカがリーダー レプリカになることを許可するかどうかに関するものです。これにより、データの損失やデータの不整合が発生する可能性があります。これは、データの一貫性に対する要件が高いシナリオ (金融分野など) では許容できない場合があります。したがって、デフォルト値は false です。ある程度のデータの不整合を許容できる場合は、これを true に設定できます。

2.4 最小同期レプリカ

ISR メカニズムのもう 1 つの関連パラメータは min.insync.replicas です。これはブローカーまたはトピック レベルで構成でき、ISR リスト内の使用可能なレプリカの最小数を表します。ここでは 2 に設定されていると仮定します。使用可能なレプリカの数がこの値より少ない場合、パーティション全体が使用できないと見なされます。このとき、クライアントがパーティションにデータを書き込むと、例外 org.apache.kafka.common.errors.NotEnoughReplicasException: 同期レプリカの数が必要な数より少ないため、メッセージが拒否されます がスローされます。

2.5 確認の送信

Kafka にはプロデューサーに対するオプションのパラメータ ack があり、これはプロデューサーがメッセージが正常に書き込まれたと判断する前にメッセージを受信する必要があるパーティション レプリカの数を指定します。

• acks=0: メッセージは正常に送信されたとみなされ、サーバーからの応答は待機されません。
• acks=1: クラスターのリーダーノードがメッセージを受信する限り、プロデューサーはサーバーから成功応答を受信します。
• acks=all: レプリケーションに参加しているすべてのノードがメッセージを受信した場合にのみ、プロデューサーはサーバーから成功応答を受信します。

3. データリクエスト

3.1 メタデータ要求メカニズム

すべてのレプリカの中で、リーダー レプリカだけがメッセージを読み書きできます。異なるパーティションのリーダー レプリカは異なるブローカー上にある可能性があるため、ブローカーがパーティション要求を受信して​​も、パーティションのリーダー レプリカがブローカー上にない場合は、クライアントに Not a Leader for Partition エラー応答が返されます。この問題を解決するために、Kafka はメタデータ要求メカニズムを提供します。

まず、クラスター内の各ブローカーは、すべてのトピックのパーティションレプリカ情報をキャッシュします。クライアントは定期的にメタデータ要求を送信し、取得したメタデータをキャッシュします。メタデータを更新する時間間隔は、クライアントの metadata.max.age.ms を構成することによって指定できます。メタデータ情報により、クライアントはリーダーレプリカが配置されているブローカーを認識し、対応するブローカーに読み取りおよび書き込み要求を直接送信します。

スケジュールされた要求の時間間隔内にパーティション レプリカの選択が行われた場合、元のキャッシュされた情報は古くなっている可能性があることを意味します。このとき、「パーティションのリーダーではありません」というエラー応答を受け取る場合もあります。この場合、クライアントはメタデータ要求を再度発行し、ローカル キャッシュを更新してから、適切なブローカーで対応する操作を実行します。プロセスは次のとおりです。

3.2 データの可視性

パーティション リーダーに保存されているすべてのデータがクライアントによって読み取られるわけではないことに注意してください。データの一貫性を確保するために、同期されたすべてのレプリカ (ISR 内のすべてのレプリカ) によって保存されたデータのみがクライアントによって読み取られます。

3.3 ゼロコピー

Kafka でのすべてのデータの書き込みと読み取りはゼロコピーを通じて実装されます。従来のコピーとゼロコピーの違いは次のとおりです。

従来のモードでは 4 つのコピーと 4 つのコンテキスト スイッチ

ディスク ファイルをネットワーク経由で送信する場合を例に挙げます。従来のモードでは、通常、次の疑似コードに示す方法を使用して、最初にファイル データをメモリに読み込み、次にメモリ内のデータをソケット経由で送信します。

バッファ = File.read  
ソケット.send(バッファ)

このプロセスには、実際には 4 つのデータのコピーが含まれます。まず、システム コールを通じてファイル データがカーネル状態バッファー (DMA コピー) に読み込まれ、次にアプリケーションがメモリ状態バッファー データをユーザー状態バッファー (CPU コピー) に読み取ります。次に、ユーザー プログラムがソケットを介してデータを送信すると、ユーザー状態バッファーのデータがカーネル状態バッファーにコピーされ (CPU コピー)、最後に DMA コピーを介して NIC バッファーにデータがコピーされます。同時に、次の図に示すように 4 つのコンテキスト スイッチが存在します。

ファイルを送信して転送するゼロコピーを実現する

Linux 2.4 以降のカーネルは、sendfile システム コールを介してゼロコピーを提供します。データは DMA 経由でカーネル バッファーにコピーされた後、CPU コピーを必要とせずに DMA 経由で NIC バッファーに直接コピーされます。これはゼロコピーという用語の由来でもあります。データのコピーが削減されるだけでなく、ファイルの読み取りとネットワーク送信全体が sendfile 呼び出しによって完了するため、プロセス全体でコンテキスト スイッチが 2 回しか発生せず、パフォーマンスが大幅に向上します。ゼロコピーのプロセスを以下の図に示します。具体的な実装の観点から見ると、Kafka のデータ転送は TransportLayer を介して完了し、そのサブクラス PlaintextTransportLayer の transferFrom メソッドは、以下に示すように、Java NIO の FileChannel の transferTo メソッドを呼び出すことによってゼロコピーを実装します。

 @オーバーライド
パブリックlong transferFrom(FileChannel fileChannel, long position, long count ) は IOException をスローします {
 fileChannel.transferTo(位置、カウント、ソケットチャネル)を返します。
 }

注意: transferTo および transferFrom は、ゼロ コピーで動作することは保証されません。実際、ゼロコピーが使用できるかどうかは、オペレーティング システムによって異なります。オペレーティング システムが sendfile などのゼロ コピー システム コールを提供している場合、これらの 2 つの方法では、そのようなシステム コールを通じてゼロ コピーの利点を最大限に活用できます。そうしないと、これら 2 つの方法自体ではゼロコピーを実現できません。

4. 物理ストレージ

4.1 パーティションの割り当て

トピックを作成するとき、Kafka はまずブローカー間でパーティションのレプリカを配布する方法を決定します。それは以下の原則に従います。

• パーティションのレプリカをすべてのブローカーに均等に分散します。
• パーティションの各レプリカが異なるブローカーに分散されていることを確認します。
• broker.rack パラメータを使用してブローカーのラック情報を指定すると、1 つのラックが使用不可になったためにパーティション全体が使用不可になることを回避するために、各パーティションのレプリカは可能な限り異なるラックのブローカーに分散されます。

上記の理由により、単一のノードに 3 つのレプリカ トピックを作成すると、通常は次の例外がスローされます。

トピック コマンドの実行中にエラーが発生しました: org.apache.kafka.common.errors.InvalidReplicationFactor
例外: レプリケーション係数: 3 が、利用可能なブローカー: 1 より大きい。

4.2 パーティションデータ保持ルール

データの保持は Kafka の基本的な機能ですが、Kafka はデータを永久に保持するわけではなく、すべてのコンシューマーがメッセージを読むまで待ってからメッセージを削除することもありません。代わりに、Kafka はトピックごとにデータ保持期間を設定し、データが削除されるまでの保持期間、または消去される前に保持されるデータの量を決定します。これらは次の 4 つのパラメータに対応します。

• log.retention.bytes : データを削除する前に許可されるデータの最大量。デフォルト値は -1 で、制限がないことを意味します。
• log.retention.ms: データ ファイルを保存するミリ秒数。設定されていない場合は、log.retention.minutes の値が使用されます。デフォルト値は null です。
• log.retention.minutes: データ ファイルを保持する分数。設定されていない場合は、log.retention.hours の値が使用されます。デフォルト値は null です。
• log.retention.hours: データ ファイルを保持する時間数。デフォルト値は 168 (1 週間) です。

大きなファイル内のメッセージの検索と削除は時間がかかり、エラーが発生しやすいため、Kafka はパーティションを複数のフラグメントに分割します。現在データを書き込んでいるフラグメントはアクティブ フラグメントと呼ばれます。アクティビティ フラグメントは削除されません。デフォルトで 1 週間データを保持し、毎日新しいフラグメントを使用する場合、新しいフラグメントが使用されるたびに最も古いフラグメントが削除されるため、ほとんどの場合、パーティションには 7 つのフラグメントが存在することになります。

4.3 ファイル形式

通常、ディスクに保存されるデータ形式は、プロデューサーによって送信されるメッセージ形式と同じです。プロデューサーが圧縮されたメッセージを送信する場合、同じバッチ内のメッセージは一緒に圧縮され、「ラップされたメッセージ」として送信され (形式は以下のとおり)、ディスクに保存されます。その後、コンシューマーはパッケージ化されたメッセージを読み取って解凍し、各メッセージの特定の情報を取得します。

まとめ

この記事では、Kafka のストレージ レプリカ メカニズムの原理とデータの保存方法について説明します。 Kafka はデータ損失を防ぐために ack メソッドを追加します。この ack は効率に多少影響する可能性があります。この ack の値はシナリオに応じて設定できます。たとえば、データが失われても問題がない場合は、0 に設定してメッセージを送信し、それを気にしなくなります。ここではビッグデータに関する情報を提供します。必要な友人は、以下の GitHub にアクセスしてダウンロードできます。自分を信じてください。あなたの努力と汗は必ず報われます。私はビッグデータブラザーです、また次回お会いしましょう~~~

この記事はWeChatの公開アカウント「ビッグデータブラザー」から転載したものです。下のQRコードからフォローできます。この記事を転載する場合はビッグデータブラザー公式アカウントまでご連絡ください。