Kafka を使用する

VPC環境で利用できます。

Apache Kafkaは高性能と安定性を保障する分散メッセージングシステムです。Kafkaは、大量のデータをリアルタイムで処理する際に使用します。

Kafkaアプリの詳細情報の確認

アプリの作成が完了すると詳細情報を確認できます。アプリの詳細情報のStatusがStableであれば、アプリが正常に起動されたことを意味します。
アプリの詳細情報を確認する方法は以下のとおりです。

1. NAVERクラウドプラットフォームコンソールで、Services > Big Data & Analytics > Data Forest > Appメニューを順にクリックします。
2. アプリを所有するアカウントを選択します。
3. 詳細情報を確認するアプリをクリックします。
4. アプリの詳細情報を確認します。
   
   • Quick links
     • AppMaster：コンテナログを確認できるURL。アプリを作成するとすべてのアプリはYARNキューに提出され、YARNは各アプリの詳細情報を確認できるウェブUIを提供。
     • kafka-manager：Kafka ManagerのURL
   • Connecting String
     • zookeeper.connect：Kafkaアプリを作成する際に指定したZookeeperアンサンブルのアドレス
   • コンポーネント
     • broker：メッセージを生産するプロデューサーと消費するコンシューマーの間でメッセージを保存、伝達する。
     • kafka-manager：brokerのモニタリング、トピック(Topic)の管理、パーティションの再割り当てができる

Kafkaアプリ使用環境の構成

起動したKafkaを使用または開発するには、まずアプリの使用環境を構成してください。DevアプリでKafkaアプリの使用環境を構成する方法について例を挙げて説明します。

1. Kafkaアプリのパッケージと設定を確認するには/home/forest/get-app-env.sh {appname} {install directory}のように作成します。

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ mkdir kafka
   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ /home/forest/get-app-env.sh kafka ./kafka/
   
   [/home/forest/get-app-env.sh] Apptype: KAFKA-2.4.0[/home/forest/get-app-env.sh] Download install-client script for KAFKA-2.4.0
   [/home/forest/get-app-env.sh] Install client on ./kafka/
   current kafka: .yarn/services/kafka/components/v1
   
   --2021-05-12 17:00:14--  http://dist.kr.df.naverncp.com/repos/release/kafka/kafka_2.12-2.4.0.tgz
   Resolving dist.kr.df.naverncp.com (dist.kr.df.naverncp.com)... 10.213.208.69
   Connecting to dist.kr.df.naverncp.com (dist.kr.df.naverncp.com)|10.213.208.69|:80... connected.
   HTTP request sent, awaiting response... 200 OK
   Length: 62283588 (59M) [application/octet-stream]
   Saving to: ‘./kafka//kafka_2.12-2.4.0.tgz’
   
   100%[============================================>] 62,283,588  --.-K/s   in 0.1s
   
   2021-05-12 17:00:14 (459 MB/s) - ‘./kafka//kafka_2.12-2.4.0.tgz’ saved [62283588/62283588]
   
   Kafka Client has been installed on ./kafka//kafka_2.12-2.4.0
   

2. インストールパスで以下のようにproducer.properties、consumer.propertiesを確認します。

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ cat ./kafka/kafka_2.12-2.4.0/config/producer.properties
   #Generated by Apache Slider
   #Wed May 12 16:48:59 KST 2021
   bootstrap.servers=broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092
   compression.type=none
   
   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ cat ./kafka/kafka_2.12-2.4.0/config/consumer.properties
   #Generated by Apache Slider
   #Wed May 12 16:48:59 KST 2021
   bootstrap.servers=broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092
   group.id=test-consumer-group
   

3. kafka-topic.shを利用して「tes」というトピックを作成します。
   kafka-topics.sh --bootstrap-server {bootstrap servers} --create --topic {topic name} --partitions {number of partitions}のように作成します。

   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ kafka/kafka_2.12-2.4.0/bin/kafka-topics.sh --bootstrap-server broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092 --create --topic test --partitions 6
   [test01@shell-0.dev.test01.kr.df.naverncp.com ~][df]$ kafka/kafka_2.12-2.4.0/bin/kafka-topics.sh --bootstrap-server broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092 --describe --topic test
   Topic: test     PartitionCount: 6       ReplicationFactor: 3    Configs: retention.bytes=50000000000
           Topic: test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
           Topic: test     Partition: 1    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
           Topic: test     Partition: 2    Leader: 2       Replicas: 2,1,0 Isr: 2,1,0
           Topic: test     Partition: 3    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0
           Topic: test     Partition: 4    Leader: 0       Replicas: 0,1,2 Isr: 0,1,2
           Topic: test     Partition: 5    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1
   

4. kafka-console-producer.shとkafka-console-consumer.shを使用してproduce/consumeする例です。

   [test01@shell-0.dev-1.test01.kr.df.naverncp.com ~][df]$ ./kafka/kafka_2.12-2.4.0/bin/kafka-console-producer.sh --broker-list broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092 --topic test
   >test1
   >test2
   >test3
   >test4                   
   >>^C
   [test01@shell-0.dev-1.test01.kr.df.naverncp.com ~][df]$ ./kafka/kafka_2.12-2.4.0/bin/kafka-console-consumer.sh --bootstrap-server broker-0.kafka.test01.kr.df.naverncp.com\:9092,broker-1.kafka.test01.kr.df.naverncp.com\:9092,broker-2.kafka.test01.kr.df.naverncp.com\:9092 --topic test --from-beginning 
   test2
   test4
   test3
   test1
   

Kafka Managerの使用

Kafka Managerを利用すると、brokerのモニタリング、トピックの管理、パーティションの再割り当てなどが行えます。

Kafka Managerの使い方を説明します。

1.NAVERクラウドプラットフォームコンソールで、Products & Services > Big Data & Analytics > Data Forest > Appメニューを順にクリックします。
2. アプリを所有するアカウントを選択してアプリをクリックします。
3. アプリの詳細情報でQuick Linksのkafka-manager URLにアクセスします。
4. ログイン画面が表示されたら、Data Forestのアカウントとパスワードを入力してログインします。
5. [Add Cluster] ボタンをクリックします。
6. Kafkaクラスタを追加します

• クラスタName：アプリ名
• クラスタZookeeper Hosts：zookeeper.connection

7. クラスタ情報を確認します。
   

Kafka brokerの変更

一つのbrokerが少ないデータを維持するように運用することをお勧めします。つまり、一つのKafkaアプリで複数のトピックを扱うよりは、一つのKafkaアプリにつき一つのトピックを扱った方が安定的です。また、トピックのデータがどれだけ増えるか予測することが難しい場合、最初からパーティションを細かく分けて、状況に応じてbroker数を調整する方式で運用することをお勧めします。

broker数を変更する方法は以下のとおりです。

1. NAVERクラウドプラットフォームコンソールで、Products & Services > Big Data & Analytics > Data Forest > Appメニューを順にクリックします。
2. アカウントを選択し、アプリを選択して [Flex] ボタンをクリックします。
3. Flexの変更画面が表示されたら、broker数を修正して [修正] ボタンをクリックします。
   

Kafkaアプリ使用時の注意事項

ローカルディスクの容量と運用方式

データはData Forestクラスタの各ノードのローカルディスクに保存されます。ただし、ノードごとのディスク容量は約300GBに制限されています。
できれば、一つのbrokerが少ないデータを維持するように運用することをお勧めします。つまり、一つのKafkaアプリで複数のトピックを扱うよりは、一つのKafkaアプリにつき一つのトピックを扱った方が安定的です。また、トピックのデータがどれだけ増えるか予測することが難しい場合、最初からパーティションを細かく分けて、状況に合わせてbroker数を調整する方式で運用することをお勧めします。

ノード障害

Kafkaアプリは、一つの物理的なノードに一つのbrokerのみ起動します。そのため、Replication Factorが3(デフォルト値)の場合、1台または2台のノード障害でもデータ消失は発生しません。

データの保存

KafkaアプリのデータはData Forestのローカルファイルシステムに保存され、ユーザーがアプリを停止(stop)するか、または他の問題によりアプリが終了した場合でもデータは削除されません。停止したアプリをユーザーが削除(destroy)する場合にのみデータを削除します。

停止したアプリを再起動(start)すると、過去に起動していたノードで起動されるため、既存データが復旧されます。しかし、このノードのリソースを他のジョブが占有しているか、ノードが障害によって外されている場合があります。その場合、1時間後にはアプリは既存ノードで起動することを諦め、他のノードで起動します。そのため、その時間に起動していないbrokerのデータは消失する可能性があります。

Zookeeperノードの整理

アプリを削除(destroy)する場合、Kafkaアプリが使用していたZookeeperパスのzookeeper.connectは削除することをお勧めします。削除せずに同じ名前のアプリを再作成(create)すると問題が生じる恐れがあります。