1. Home
  2. Platform
  3. 管理
  4. Kafka の運用
  5. データの自動バランス調整

チュートリアル: データの自動バランス調整（Docker）

注釈

Confluent Platform 6.0.0 以降では Self-Balancing Clusters が提供されており、Auto Data Balancer の代わりに使用することが推奨されています。詳細な機能の比較については、「Self-Balancing と Auto Data Balancer」を参照してください。Self-Balancing Docker クイックスタートについては、「Self-Balancing Clusters デモ（Docker）」を参照してください。

このチュートリアルでは Confluent Auto Data Balancer （ADB）を Confluent Server で実行します。これによりデータを移行し、クラスター全体でワークロードを均等にすることができます。このチュートリアルを終了すると、Auto Data Balancer CLI ツールを正しく実行して、ブローカーの追加または削除後にデータのバランスを調整できるようになります。

注釈

このチュートリアルでは、データを Docker コンテナーにローカルに格納するように、Kafka と ZooKeeper を構成します。本稼働環境でのデプロイ（つまり一般にデータ紛失が常に重要な問題である場合）では、コンテナーが動作を停止した場合や再起動した場合にデータを維持するために、マウントされたボリュームを使用する必要があります。Docker で Kafka のようなシステムを実行している場合、ファイルシステムはメッセージの格納やキャッシュに大きく依存しているため、この点は重要です。マウントされたボリュームをホストマシンに追加する方法の例については、 Docker 外部ボリュームに関するドキュメント を参照してください。

Docker のインストールと実行

このチュートリアルでは、Docker クライアントを使用して Docker を実行します。イメージ実行用の Docker Compose の使用に関する詳細については、 こちらのガイドの最後の部分 を参照してください。

開始するには、Docker をインストールし、実行します 。Confluent Platform Docker イメージには Docker バージョン 1.11 以上 が必要です。

前提条件

Docker クライアント : 3 ノード Kafka クラスターのセットアップ

注釈

次の手順では、それぞれの Docker コンテナーがデタッチモードで動作します。実行中のコンテナーに関するログにアクセスする方法を確認できます。-d フラグを -it で置き換えることで、コンテナーをフォアグラウンドで実行することもできます。

  1. Git リポジトリにクローンを作成して、サンプルのディレクトリに移動します。

    git clone git@github.com:confluentinc/kafka-images.git
    Copy
    cd kafka-images/examples/enterprise-kafka
    Copy

  2. サンプルの Docker Compose ファイルを使用して各サービスを開始します。Docker Compose ファイルには、1 つの ZooKeeper および 6 つの Kafka ブローカー用の構成プロパティがあります。これらのブローカーは 2 つのラックに配置されるように構成されます。3 つのブローカーがある 1 つのラックが開始されると、サンプルデータを含むトピックが作成され、クラスターのバランスを調整するために Auto Data Balancer CLI ツールが実行されます。この手順を終了すると、ブローカーのもう 1 つのラックを追加して Auto Data Balancer CLI ツールを実行し、新たに追加されたブローカーの間でデータのバランスを調整できるようになります。

    1. ZooKeeper およびブローカーの最初のラックを Docker Compose コマンドで開始します。

      docker-compose create
      Copy

      次のように表示されます。

      Creating enterprise-kafka_zookeeper_1 ... done
Creating enterprise-kafka_kafka-1_1   ... done
Creating enterprise-kafka_kafka-2_1   ... done
Creating enterprise-kafka_kafka-3_1   ... done
Creating enterprise-kafka_kafka-4_1   ... done
Creating enterprise-kafka_kafka-5_1   ... done
Creating enterprise-kafka_kafka-6_1   ... done
      Copy

    2. 各サービスを起動します。

      docker-compose start zookeeper kafka-1 kafka-2 kafka-3
      Copy

      次のように表示されます。

      Starting zookeeper ... done
Starting kafka-1   ... done
Starting kafka-2   ... done
Starting kafka-3   ... done
      Copy

    3. これらはオプションの検証手順です。

      • 次のコマンドと使用して、各サービスが実行されているかを確認します。

        docker-compose ps
        Copy

        次のように表示されます。

          Name                        Command            State    Ports
------------------------------------------------------------------------
enterprisekafka_kafka-1_1     /etc/confluent/docker/run   Up
enterprisekafka_kafka-2_1     /etc/confluent/docker/run   Up
enterprisekafka_kafka-3_1     /etc/confluent/docker/run   Up
enterprisekafka_kafka-4_1     /etc/confluent/docker/run   Exit 0
enterprisekafka_kafka-5_1     /etc/confluent/docker/run   Exit 0
enterprisekafka_kafka-6_1     /etc/confluent/docker/run   Exit 0
enterprisekafka_zookeeper_1   /etc/confluent/docker/run   Up
        Copy

      • ZooKeeper が正常であるかどうかを ZooKeeper ログをチェックして確認します。

        docker-compose logs zookeeper | grep -i binding
        Copy

        ターミナルウィンドウに次のように表示されます。

        zookeeper_1  | [2016-10-21 22:15:22,494] INFO binding to port 0.0.0.0/0.0.0.0:22181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
        Copy

      • Kafka ログで送信先クラスターをチェックして、そのブローカーが正常であることを確認します。

        docker-compose logs kafka-1 | grep -i started
        Copy

        次のようなメッセージが表示されます。

        kafka-1_1    | [2016-10-21 22:19:50,964] INFO [Socket Server on Broker 1], Started 1 acceptor threads (kafka.network.SocketServer)
kafka-1_1    | [2016-10-21 22:19:51,300] INFO [Kafka Server 1], started (kafka.server.KafkaServer)
....
        Copy

  3. ブローカーが起動しました。次に、adb-test というテストトピックを作成します。

    docker run \
  --net=host \
  --rm confluentinc/cp-kafka:6.1.5 \
  kafka-topics --create --topic adb-test --partitions 20 --replication-factor 3 --if-not-exists --bootstrap-server localhost:19092
    Copy

    ターミナルウィンドウに次のような出力が表示されます。

    Created topic "adb-test".
    Copy

    省略可能: トピックが正常に作成されたことを確認します。

    docker run \
  --net=host \
  --rm confluentinc/cp-kafka:6.1.5 \
  kafka-topics --describe --topic adb-test --bootstrap-server localhost:19092
    Copy

    ターミナルウィンドウに次のような出力が表示されます。

    Topic:adb-test      PartitionCount:20       ReplicationFactor:3     Configs:
Topic: adb-test     Partition: 0    Leader: 2       Replicas: 2,1,3 Isr: 2,1,3
Topic: adb-test     Partition: 1    Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
Topic: adb-test     Partition: 2    Leader: 1       Replicas: 1,3,2 Isr: 1,3,2
Topic: adb-test     Partition: 3    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
Topic: adb-test     Partition: 4    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
Topic: adb-test     Partition: 5    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
Topic: adb-test     Partition: 6    Leader: 2       Replicas: 2,1,3 Isr: 2,1,3
Topic: adb-test     Partition: 7    Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
Topic: adb-test     Partition: 8    Leader: 1       Replicas: 1,3,2 Isr: 1,3,2
Topic: adb-test     Partition: 9    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
Topic: adb-test     Partition: 10   Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
Topic: adb-test     Partition: 11   Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
Topic: adb-test     Partition: 12   Leader: 2       Replicas: 2,1,3 Isr: 2,1,3
Topic: adb-test     Partition: 13   Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
Topic: adb-test     Partition: 14   Leader: 1       Replicas: 1,3,2 Isr: 1,3,2
Topic: adb-test     Partition: 15   Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
Topic: adb-test     Partition: 16   Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
Topic: adb-test     Partition: 17   Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
Topic: adb-test     Partition: 18   Leader: 2       Replicas: 2,1,3 Isr: 2,1,3
Topic: adb-test     Partition: 19   Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
    Copy

  4. 次に、データを新しいトピックに生成できるかどうかを試します。

    docker run \
  --net=host \
  --rm \
  confluentinc/cp-kafka:6.1.5 \
  bash -c 'kafka-producer-perf-test --topic adb-test --num-records 2000000 --record-size 1000 --throughput 100000 --producer-props bootstrap.servers=localhost:19092'
    Copy

    このコマンドでは組み込みの Kafka Performance Producer を使用し、2 GB のサンプルデータをトピックに生成します。実行すると、次のような結果が表示されます。

    209047 records sent, 41784.3 records/sec (39.85 MB/sec), 91.1 ms avg latency, 520.0 max latency.
325504 records sent, 65100.8 records/sec (62.08 MB/sec), 35.6 ms avg latency, 474.0 max latency.
258023 records sent, 51573.7 records/sec (49.18 MB/sec), 359.6 ms avg latency, 1264.0 max latency.
287934 records sent, 57586.8 records/sec (54.92 MB/sec), 455.1 ms avg latency, 1429.0 max latency.
413091 records sent, 81978.8 records/sec (78.18 MB/sec), 200.6 ms avg latency, 757.0 max latency.
282214 records sent, 56128.5 records/sec (53.53 MB/sec), 495.6 ms avg latency, 1738.0 max latency.
85071 records sent, 16815.8 records/sec (16.04 MB/sec), 468.0 ms avg latency, 3861.0 max latency.
115 records sent, 8.8 records/sec (0.01 MB/sec), 8307.4 ms avg latency, 13127.0 max latency.
13358 records sent, 2671.6 records/sec (2.55 MB/sec), 15408.9 ms avg latency, 23005.0 max latency.
74948 records sent, 14284.0 records/sec (13.62 MB/sec), 6555.0 ms avg latency, 22782.0 max latency.
5052 records sent, 1010.4 records/sec (0.96 MB/sec), 3228.3 ms avg latency, 8508.0 max latency.
2000000 records sent, 30452.988199 records/sec (29.04 MB/sec), 786.61 ms avg latency, 23005.00 ms max latency, 82 ms 50th, 1535 ms 95th, 22539 ms 99th, 22929 ms 99.9th.
    Copy

  5. confluent-rebalancer を実行して、クラスターでデータのバランスを調整します。

    docker run \
  --net=host \
  --rm \
  confluentinc/cp-enterprise-kafka:6.1.5 \
  bash -c "confluent-rebalancer execute --bootstrap-server localhost:19092 --metrics-bootstrap-server localhost:19092 --throttle 100000000 --force --verbose"
    Copy

    バランス調整が開始され、次のような結果が表示されます。

    You are about to move 6 replica(s) for 6 partitions to 1 broker(s) with total size 0.9 MB.
The preferred leader for 6 partition(s) will be changed.
In total, the assignment for 7 partitions will be changed.

The following brokers will require more disk space during the rebalance and, in some cases, after the rebalance:
    Broker     Current (MB)    During Rebalance (MB)  After Rebalance (MB)
    2          2,212.8         2,213.8                2,213.8

Min/max stats for brokers (before -> after):
      Type  Leader Count                 Replica Count                Size (MB)
      Min   8 (id: 2) -> 10 (id: 1)      21 (id: 2) -> 27 (id: 1)     2,069.6 (id: 1) -> 2,069.1 (id: 1)
      Max   12 (id: 3) -> 11 (id: 2)     30 (id: 1) -> 27 (id: 1)     2,212.8 (id: 2) -> 2,213.8 (id: 2)

Rack stats (before -> after):
      Rack       Leader Count    Replica Count   Size (MB)
      rack-a     31 -> 31        81 -> 81        6,352 -> 6,352

Broker stats (before -> after):
      Broker     Leader Count    Replica Count   Size (MB)
      1          11 -> 10        30 -> 27        2,069.6 -> 2,069.1
      2          8 -> 11         21 -> 27        2,212.8 -> 2,213.8
      3          12 -> 10        30 -> 27        2,069.6 -> 2,069.1

The rebalance has been started, run ``status`` to check progress.

Warning: You must run the ``status`` or ``finish`` command periodically, until the rebalance completes, to ensure the throttle is removed. You can also alter the throttle by re-running the execute command passing a new value.
    Copy

    次のコマンドを実行して、バランス調整処理のステータスをチェックできます。

    docker run \
  --net=host \
  --rm \
  confluentinc/cp-enterprise-kafka:6.1.5 \
  bash -c "confluent-rebalancer status --bootstrap-server localhost:19092"
    Copy

    7 partitions are being rebalanced などのメッセージが表示されている場合、15 ～ 20 秒ほど待機し、No rebalance is currently in progress が表示されるまで前に示したコマンドを再実行します。これはバランス調整処理が正常に完了したことを意味します。

    次のコマンド（確実にレプリケーションスロットルを削除）を実行して、バランス調整処理を終了できます。

    docker run \
  --net=host \
  --rm \
  confluentinc/cp-enterprise-kafka:6.1.5 \
  bash -c "confluent-rebalancer finish --bootstrap-server localhost:19092"
    Copy

    ログに次の内容が表示されます。

    The rebalance has completed and throttling has been disabled
    Copy

  6. Auto Data Balancer を使用すると、新しいブローカーをクラスターに簡単に追加できます。ここで新しいラックを丸ごとクラスターに追加し、バランス調整処理を再実行して、クラスター全体でデータのバランスを調整できます。

    次のコマンドを実行して新しいラックを開始します。

    docker-compose start kafka-4 kafka-5 kafka-6
    Copy

    ステップ 4 の手順に従って、Kafka ブローカーが健全であることを確認します。

    手順 5 に従って、バランス調整処理を開始します。バランス調整処理が終了した後には、クラスター全体でデータのバランスが調整されています。次のようにトピックメタデータを記述して、これを確認します。

    docker run \
  --net=host \
  --rm confluentinc/cp-kafka:6.1.5 \
  kafka-topics --describe --topic adb-test --bootstrap-server localhost:19092
    Copy

    すべてのブローカーにパーティションが分散されていることがわかります（いくつかのレプリカとリーダーがブローカー 4、5 または 6 に割り当てられています）。

    Topic:adb-test      PartitionCount:20       ReplicationFactor:3     Configs:
Topic: adb-test     Partition: 0    Leader: 1       Replicas: 1,5,6 Isr: 5,1,6
Topic: adb-test     Partition: 1    Leader: 3       Replicas: 3,5,4 Isr: 5,3,4
Topic: adb-test     Partition: 2    Leader: 6       Replicas: 6,4,1 Isr: 1,6,4
Topic: adb-test     Partition: 3    Leader: 6       Replicas: 6,5,3 Isr: 5,6,3
Topic: adb-test     Partition: 4    Leader: 1       Replicas: 1,4,5 Isr: 5,1,4
Topic: adb-test     Partition: 5    Leader: 3       Replicas: 6,4,3 Isr: 6,3,4
Topic: adb-test     Partition: 6    Leader: 1       Replicas: 5,1,6 Isr: 5,1,6
Topic: adb-test     Partition: 7    Leader: 3       Replicas: 3,5,4 Isr: 5,3,4
Topic: adb-test     Partition: 8    Leader: 4       Replicas: 4,6,1 Isr: 1,6,4
Topic: adb-test     Partition: 9    Leader: 5       Replicas: 5,6,3 Isr: 5,6,3
Topic: adb-test     Partition: 10   Leader: 2       Replicas: 2,4,5 Isr: 5,2,4
Topic: adb-test     Partition: 11   Leader: 4       Replicas: 4,2,6 Isr: 6,2,4
Topic: adb-test     Partition: 12   Leader: 5       Replicas: 5,2,6 Isr: 5,6,2
Topic: adb-test     Partition: 13   Leader: 2       Replicas: 2,5,4 Isr: 5,2,4
Topic: adb-test     Partition: 14   Leader: 4       Replicas: 4,6,2 Isr: 6,2,4
Topic: adb-test     Partition: 15   Leader: 1       Replicas: 1,3,2 Isr: 1,2,3
Topic: adb-test     Partition: 16   Leader: 2       Replicas: 3,2,1 Isr: 2,1,3
Topic: adb-test     Partition: 17   Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
Topic: adb-test     Partition: 18   Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
Topic: adb-test     Partition: 19   Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
    Copy

  7. ブローカーを削除し、バランス調整処理を再び実行してみます。

    ヒント: バランス調整計画からブローカーを除外することをリバランサーに通知する必要があります。たとえば、ブローカー 1 を削除するには、以下のコマンドを実行します。

    docker run \
  --net=host \
  --rm \
  confluentinc/cp-enterprise-kafka:6.1.5 \
  bash -c "confluent-rebalancer execute --bootstrap-server localhost:19092 --metrics-bootstrap-server localhost:19092 --throttle 100000000 --force --verbose --remove-broker-ids 1"
    Copy

  8. confluent-rebalance コマンドを自由に試してみてください。完了したら、次のコマンドですべてのコンポーネントをシャットダウンします。

    docker-compose stop
    Copy

    すべてのコンテナーを削除する場合は、次のとおり実行します。

    docker-compose rm
    Copy