Kafka の運用とメンテナンス |データ移行を本当に理解していますか?

[[420468]]

この記事では以下の内容を説明します

[Kafka 運用保守] レプリカの拡張と縮小、データ移行、レプリカの再配布、レプリカのクロスパス移行

スクリプトパラメータ

1. スクリプトの使用方法の紹介

このスクリプトは、パーティションを再配布するために Kafka によって提供されるスクリプト ツールです。

1.1 推奨設定スクリプトを生成する

キーパラメータ --generate

パーティションのレプリカを再配布する前に、次の方法で適切な割り当てファイルを取得することをお勧めします。 move-json-file.json ファイルを書き込みます。このファイルは、どのトピックを再配布するかの計算を示します

{
 「トピック」 : [
    { "トピック" : "test_create_topic1" }
  ]、
 「バージョン」 : 1
 }

次に、次のスクリプトを実行します。--broker-list "0,1,2,3" はブローカーに割り当てるパラメーターです。

 sh bin/kafka-reassign-partitions.sh --zookeeper xxx:2181 --topics-to-move-json-file config/move-json-file.json --broker-list "0,1,2,3" --generate  

実行後、次のように印刷されます。

現在のパーティションのレプリカの割り当て//現在のレプリカの割り当て方法
{ "バージョン" :1、 "パーティション" :[{ "トピック" : "test_create_topic1" 、 "パーティション" :2、 "レプリカ" :[1]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :1、 "レプリカ" :[3]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :0、 "レプリカ" :[2]、 "log_dirs" :[ "任意" ]}]} 
 
提案されたパーティション再割り当て構成//予想される再配分方法
{ "バージョン" :1、 "パーティション" :[{ "トピック" : "test_create_topic1" 、 "パーティション" :2、 "レプリカ" :[2]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :1、 "レプリカ" :[1]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :0、 "レプリカ" :[0]、 "log_dirs" :[ "任意" ]}]}

パーティションの移動はまだ開始されておらず、現在の割り当てと提案が通知されるだけであることに注意してください。元に戻したい場合に備えて、現在の割り当てを保存します

1.2. Jsonファイルを実行する

キーパラメータ --execute 上記で取得した予想される再配布方法ファイルをjsonファイルreassignment-json-file.jsonに保存します。

 { "バージョン" :1、 "パーティション" :[{ "トピック" : "test_create_topic1" 、 "パーティション" :2、 "レプリカ" :[2]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :1、 "レプリカ" :[1]、 "log_dirs" :[ "任意" ]}、{ "トピック" : "test_create_topic1" 、 "パーティション" :0、 "レプリカ" :[0]、 "log_dirs" :[ "任意" ]}]}

次に実行する

sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute  

移行プロセス中は、クラスター上のトラフィックの急増による影響に注意してください。 Kafka はブローカー間のレプリケーション転送にトラフィック制限を提供し、これにより、あるマシンから別のマシンへのレプリカの帯域幅の上限が制限されます。これは、クラスターの再バランス調整、新しいブローカーの起動、ブローカーの追加または削除を行うときに非常に便利です。ユーザーへの影響を確実にするために、これらの集中的なデータ操作を制限するため、たとえば、上記の移行操作にフロー制限オプション（スロットル50000000）を追加します。

 > sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute -- throttle 50000000  

最後に --throttle 50000000 パラメータを追加すると、モバイル パーティションを実行するときに、フローが 50000000 B/s に制限されます。パラメータを追加すると、

スロットル制限が設定された  50000000 B/sまで
パーティションの再割り当てが正常に開始されました。

移行中にレプリカのクロスパス移行（同じブローカーの複数のパス）を含める場合、このフロー制限対策は有効にならないことに注意してください。同じブローカーの異なるパスの直接移行のフロー制限を制限するフロー制限パラメータ |--replica-alter-log-dirs-throttle を追加する必要があります。

再バランス調整中に制限を変更する場合は、スループットを増やして、より速く完了するようにします。同じreassignment-json-fileでexecuteコマンドを再実行することができます。

1.3.検証

主要なパラメータ --verify このオプションはパーティションの再割り当てのステータスを確認するために使用され、 --throttle トラフィック制限も削除されます。そうしないと、通常のレプリケーション操作のトラフィックも制限される可能性があります。

sh bin/kafka-reassign-partitions.sh --zookeeper xxxx:2181 --reassignment-json-file config/reassignment-json-file.json --verifyImage注: 入力した BrokerId が存在しない場合は、レプリカの操作は失敗しますが、他のレプリカには影響しません。例えば画像では

2. レプリカのスケーリング

Kafka は、パーティションを拡張できる kafka-topic.sh スクリプトとは異なり、レプリカの拡張と縮小をサポートする専用のスクリプトを提供しません。レプリカを拡大したり縮小したりしたい場合は、回りくどい方法でしか実行できません。 kafka-reassign-partitions.sh を使用してレプリカを再割り当てする

2.1 レプリ​​カの拡張

現在の状況は 3 つのパーティションと 1 つのレプリカであると仮定します。可用性を向上させるために、レプリカの数を 2 に増やしたいと考えています。

2.1.1 計算レプリカ割り当て方法

ステップ1.1で--generateを使用して現在の割り当てを取得し、次のjsonを取得します。

 {
 「バージョン」 : 1,
 「パーティション」 : [{
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 2,
 「レプリカ」 ：[2]、
 "log_dirs" : [ "任意" ]
    }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 1,
 「レプリカ」 ：[1]、
 "log_dirs" : [ "任意" ]
    }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 0,
 「レプリカ」 : [0],
 "log_dirs" : [ "任意" ]
    }]
 }

すべてのパーティションのレプリカ数を 2 に変更したいので、ブローカーのリストである "replicas": [] の値を変更するだけで済みます。最初のブローカーはリーダーです。そこで、次のように、割り当てルールに従ってjsonファイルを変更します。

 {
 「バージョン」 : 1,
 「パーティション」 : [{
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 2,
 「レプリカ」 : [2,0],
 "log_dirs" : [ "任意" , "任意" ]
    }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 1,
 「レプリカ」 : [1,2],
 "log_dirs" : [ "任意" , "任意" ]
    }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 0,
 「レプリカ」 : [0,1],
 "log_dirs" : [ "任意" , "任意" ]
    }]
 }

log_dirs の数はレプリカの数と一致する必要があることに注意してください。または、単に log_dirs オプションを削除します。このlog_dirsは、レプリカがパス間で移行される場合の絶対パスです。

2.1.2 実行

再バランス調整中に制限を変更する場合は、スループットを増やして、より速く完了するようにします。同じ reassignment-json-file を使用して実行コマンドを再実行できます。

2.1.2 検証

絵が完成すると、コピーの数が増えます。

2.2 レプリカのスケーリング

レプリカのスケーリングはスケールアップと同じです。割り当てられたレプリカの数が以前の数より少ない場合、余分なレプリカは削除されます。たとえば、レプリカを追加したばかりで、それをレプリカに復元したいとします。

次のjsonファイルを実行します

{
 「バージョン」 : 1,
 「パーティション」 : [{
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 2,
 「レプリカ」 ：[2]、
 "log_dirs" : [ "任意" ]
  }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 1,
 「レプリカ」 ：[1]、
 "log_dirs" : [ "任意" ]
  }, {
 「トピック」 : 「test_create_topic1」 、
 「パーティション」 : 0,
 「レプリカ」 : [0],
 "log_dirs" : [ "任意" ]
  }]
 }

実行後、他のコピーが削除対象としてマークされていることがわかります。しばらくするとクリーンアップされます。

このようにレプリカの拡張と縮小は実現できましたが、負荷分散などレプリカの分散は自分で制御する必要があります。Kafka が自動的に分散を手伝ってくれるよりも断然合理的です。では、合理的に分散された Json ファイルを提供するのに役立つ良い方法はあるでしょうか?追伸:

以前、Kafka ソース コードでトピックを作成するときのパーティションとレプリカの割り当てルールを分析しました。では、このような割り当てプロセスを割り当てる際に、同じルールを使用しても問題ないのではないでしょうか。 --generate は基本的にこのメソッドを呼び出します、AdminUtils.assignReplicasToBrokers(brokerMetadatas, assignment.size, replicas.size)

具体的な実装操作については【Kafkaの思考】最小コスト拡張・縮小レプリカ設計計画を参照

同様の方法を実装するプロジェクトを作成します。面倒な場合は、LogIKM の新しいコピー機能を使用してこれを直接実行できます。 （今後実装予定）

3. パーティションの拡張

Kafka パーティション拡張は kafka-topis.sh スクリプトによって実装されます。縮小とパーティション拡張をサポートしていない場合は、[kafka ソースコード] トピックコマンド alter ソースコード分析 (パーティション拡張) を参照してください。

4. パーティションの移行

パーティションの移行は上記と同じです。1.1、1.2、1.3 を参照してください。

5. レプリカのクロスパス移行

オンライン Kafka マシンの各ディスクの占有率が不均一で、一方的になることが多いのはなぜですか?これは、Kafka は各ディスク上のパーティション数が均等に分散されていることを保証するだけで、各パーティションが実際に占有するスペースを把握できないためです。したがって、一部のパーティション内のメッセージ数が膨大になり、大量のディスク領域が占有される可能性が非常に高くなります。バージョン 1.1 より前では、Kafka は異なるブローカー間でのパーティション化されたデータの再配布のみをサポートしており、同じブローカー上の異なるディスク間でのデータの再配布はできなかったため、ユーザーにはこの問題を解決する方法がありませんでした。バージョン1.1では、異なるパス間でのレプリカの移行が正式にサポートされています。

複数のパスを使用してブローカーにパーティションを保存するにはどうすればよいですか?

複数のフォルダーを構成に接続するだけです。

 ############################# ログの基本 ############################ 
 
 #ログファイルを保存するディレクトリのカンマ区切りリスト
log.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8

同じブローカー上の異なるパスには異なるパーティションのみが格納され、レプリカは同じブローカーには格納されないことに注意してください。そうしないとレプリカは無意味になります（災害復旧）

クロスパス移行にはどのように対処しますか?

移行 json ファイルには log_dirs というパラメーターがあります。デフォルトのリクエストが通過しない場合は、"log_dirs": ["any"] です (この配列の数はコピーと一致している必要があります) ただし、パス間の移行を実現する場合は、次のように絶対パスを入力するだけで済みます。

移行jsonファイルの例

{
 「バージョン」 : 1,
 「パーティション」 : [{
 「トピック」 : 「test_create_topic4」 、
 「パーティション」 : 2,
 「レプリカ」 : [0],
 "log_dirs" : [ "/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-5" ]
  }, {
 「トピック」 : 「test_create_topic4」 、
 「パーティション」 : 1,
 「レプリカ」 : [0],
 "log_dirs" : [ "/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-6" ]
  }]
 }

次にスクリプトを実行します

sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx --reassignment-json-file config/reassignment-json-file.json --execute --bootstrap-server  
 xxxxx:9092 --レプリカ-変更-ログ-ディレクトリ-スロットル 10000  

注意: パス間で移行する場合は、--bootstrap-server を渡す必要があります。

フローを制限する必要がある場合は、パラメータ |--replica-alter-log-dirs-throttle; を追加します。 --throttle とは異なり、--replica-alter-log-dirs-throttle は Broker 内の異なるパスの移行フローを制限します。

ソースコード分析

ソースコード分析については、[kafka ソースコード] ReassignPartitionsCommand ソースコード分析 (レプリカの拡張と縮小、データ移行、パーティションの再配分、レプリカのクロスパス移行) の記事を参照してください。

ビデオリンク: https://mp.weixin.qq.com/s/fQ03wpctV1dGnmk1r-xEWA

この記事はWeChat公式アカウント「Shi Zhenzhen's Grocery Store」から転載したものです。下のQRコードからフォローできます。記事を転載する場合は、石真珍食料品店の公式アカウントまでご連絡ください。