CentOS7にApacheKafkaをインストールする方法
著者は、 Write for DOnations プログラムの一環として、 Free and Open SourceFundを選択して寄付を受け取りました。
序章
Apache Kafka は、大量のリアルタイムデータを効率的に処理するように設計された人気の分散メッセージブローカーです。 Kafkaクラスターは、拡張性とフォールトトレラント性が高いだけでなく、ActiveMQやRabbitMQなどの他のメッセージブローカーと比較してスループットがはるかに高くなっています。 一般にパブリッシュ/サブスクライブメッセージングシステムとして使用されますが、パブリッシュされたメッセージの永続的なストレージを提供するため、多くの組織でもログ集約に使用されます。
パブリッシュ/サブスクライブメッセージングシステムを使用すると、1つ以上のプロデューサーが、コンシューマーの数やメッセージの処理方法を考慮せずにメッセージをパブリッシュできます。 購読しているクライアントには、更新と新しいメッセージの作成について自動的に通知されます。 このシステムは、クライアントが定期的にポーリングして新しいメッセージが利用可能かどうかを判断するシステムよりも効率的でスケーラブルです。
このチュートリアルでは、CentOS7にApacheKafka2.1.1をインストールして使用します。
前提条件
フォローするには、次のものが必要です。
- 1台のCentOS7サーバーとsudo権限を持つroot以外のユーザー。 root以外のユーザーを設定していない場合は、このガイドで指定されている手順に従ってください。
- サーバー上に少なくとも4GBのRAM。 この量のRAMを使用せずにインストールすると、Kafkaサービスが失敗し、 Java仮想マシン(JVM)が起動時に「メモリ不足」例外をスローする可能性があります。
- OpenJDK8がサーバーにインストールされています。 このバージョンをインストールするには、OpenJDKの特定のバージョンをインストールする際のこれらの手順に従ってください。 KafkaはJavaで記述されているため、JVMが必要です。 ただし、その起動シェルスクリプトにはバージョン検出のバグがあり、8を超えるJVMバージョンでは起動に失敗します。
ステップ1—Kafkaのユーザーを作成する
Kafkaはネットワーク経由でリクエストを処理できるため、専用のユーザーを作成する必要があります。 これにより、Kafkaサーバーが危険にさらされた場合のCentOSマシンへの損傷が最小限に抑えられます。 このステップでは専用のkafkaユーザーを作成しますが、Kafkaのセットアップが完了したら、このサーバーで他のタスクを実行するために別の非rootユーザーを作成する必要があります。
root以外のsudoユーザーとしてログインし、useradd
コマンドを使用してkafkaというユーザーを作成します。
sudo useradd kafka -m
-m
フラグは、ユーザーのホームディレクトリが作成されることを保証します。 このホームディレクトリ/home/kafka
は、以下のセクションのコマンドを実行するためのワークスペースディレクトリとして機能します。
passwd
を使用してパスワードを設定します。
sudo passwd kafka
kafkaユーザーをadduser
コマンドを使用してwheel
グループに追加し、Kafkaの依存関係をインストールするために必要な権限を付与します。
sudo usermod -aG wheel kafka
これで、kafkaユーザーの準備が整いました。 su
を使用してこのアカウントにログインします。
su -l kafka
Kafka固有のユーザーを作成したので、Kafkaバイナリのダウンロードと抽出に進むことができます。
ステップ2—Kafkaバイナリのダウンロードと抽出
Kafkaバイナリをダウンロードして、kafkaユーザーのホームディレクトリの専用フォルダに抽出してみましょう。
まず、/home/kafka
にDownloads
というディレクトリを作成してダウンロードを保存します。
mkdir ~/Downloads
curl
を使用して、Kafkaバイナリをダウンロードします。
curl "https://www.apache.org/dist/kafka/2.1.1/kafka_2.11-2.1.1.tgz" -o ~/Downloads/kafka.tgz
kafka
というディレクトリを作成し、このディレクトリに移動します。 これは、Kafkaインストールのベースディレクトリになります。
mkdir ~/kafka && cd ~/kafka
tar
コマンドを使用してダウンロードしたアーカイブを抽出します。
tar -xvzf ~/Downloads/kafka.tgz --strip 1
--strip 1
フラグを指定して、アーカイブの内容が~/kafka/
自体に抽出され、アーカイブ内の別のディレクトリ(~/kafka/kafka_2.11-2.1.1/
など)には抽出されないようにします。
バイナリのダウンロードと抽出が正常に行われたので、トピックの削除を可能にするためにKafkaへの構成に進むことができます。
ステップ3—Kafkaサーバーの構成
Kafkaのデフォルトの動作では、トピック、メッセージを公開できるカテゴリ、グループ、またはフィード名を削除できません。 これを変更するには、構成ファイルを編集しましょう。
Kafkaの構成オプションは、server.properties
で指定されています。 vi
またはお気に入りのエディターでこのファイルを開きます。
vi ~/kafka/config/server.properties
Kafkaトピックを削除できるようにする設定を追加しましょう。 i
を押してテキストを挿入し、ファイルの最後に以下を追加します。
〜/ kafka / config / server.properties
delete.topic.enable = true
終了したら、ESC
を押して挿入モードを終了し、:wq
を押してファイルに変更を書き込んで終了します。 Kafkaを構成したので、実行用のsystemdユニットファイルの作成と起動時の有効化に進むことができます。
ステップ4—Systemdユニットファイルの作成とKafkaサーバーの起動
このセクションでは、Kafkaサービス用のsystemdユニットファイルを作成します。 これは、他のLinuxサービスと一貫した方法で、Kafkaの開始、停止、再起動などの一般的なサービスアクションを実行するのに役立ちます。
Zookeeperは、Kafkaがクラスターの状態と構成を管理するために使用するサービスです。 これは、多くの分散システムで不可欠なコンポーネントとして一般的に使用されています。 詳細については、公式の Zookeeperdocsにアクセスしてください。
zookeeper
のユニットファイルを作成します。
sudo vi /etc/systemd/system/zookeeper.service
次のユニット定義をファイルに入力します。
/etc/systemd/system/zookeeper.service
[Unit] Requires=network.target remote-fs.target After=network.target remote-fs.target [Service] Type=simple User=kafka ExecStart=/home/kafka/kafka/bin/zookeeper-server-start.sh /home/kafka/kafka/config/zookeeper.properties ExecStop=/home/kafka/kafka/bin/zookeeper-server-stop.sh Restart=on-abnormal [Install] WantedBy=multi-user.target
[Unit]
セクションでは、Zookeeperを起動する前に、ネットワークとファイルシステムの準備ができている必要があることを指定しています。
[Service]
セクションでは、systemdがサービスの開始と停止にzookeeper-server-start.sh
およびzookeeper-server-stop.sh
シェルファイルを使用する必要があることを指定しています。 また、Zookeeperが異常終了した場合に自動的に再起動する必要があることも指定しています。
編集が終了したら、ファイルを保存して閉じます。
次に、kafka
のsystemdサービスファイルを作成します。
sudo vi /etc/systemd/system/kafka.service
次のユニット定義をファイルに入力します。
/etc/systemd/system/kafka.service
[Unit] Requires=zookeeper.service After=zookeeper.service [Service] Type=simple User=kafka ExecStart=/bin/sh -c '/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties > /home/kafka/kafka/kafka.log 2>&1' ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh Restart=on-abnormal [Install] WantedBy=multi-user.target
[Unit]
セクションは、このユニットファイルがzookeeper.service
に依存することを指定します。 これにより、kafa
サービスの開始時にzookeeper
が自動的に開始されます。
[Service]
セクションでは、systemdがサービスの開始と停止にkafka-server-start.sh
およびkafka-server-stop.sh
シェルファイルを使用する必要があることを指定しています。 また、Kafkaが異常終了した場合に自動的に再起動する必要があることも指定しています。
編集が終了したら、ファイルを保存して閉じます。
ユニットが定義されたので、次のコマンドでKafkaを起動します。
sudo systemctl start kafka
サーバーが正常に起動したことを確認するには、kafka
ユニットのジャーナルログを確認してください。
journalctl -u kafka
次のような出力が表示されます。
OutputJul 17 18:38:59 kafka-centos systemd[1]: Started kafka.service.
これで、ポート9092
でリッスンしているKafkaサーバーができました。
kafka
サービスを開始しているときに、サーバーを再起動した場合、サーバーは自動的に開始されません。 サーバーの起動時にkafka
を有効にするには、次のコマンドを実行します。
sudo systemctl enable kafka
サービスを開始して有効にしたので、インストールを確認しましょう。
ステップ5—インストールのテスト
「HelloWorld」メッセージを公開して使用し、Kafkaサーバーが正しく動作していることを確認しましょう。 Kafkaでメッセージを公開するには、次のものが必要です。
- プロデューサー。トピックへのレコードとデータの公開を可能にします。
- コンシューマー。トピックからメッセージとデータを読み取ります。
まず、次のように入力して、TutorialTopic
という名前のトピックを作成します。
~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TutorialTopic
次の出力が表示されます。
OutputCreated topic "TutorialTopic".
kafka-console-producer.sh
スクリプトを使用して、コマンドラインからプロデューサーを作成できます。 Kafkaサーバーのホスト名、ポート、およびトピック名を引数として想定しています。
次のように入力して、文字列"Hello, World"
をTutorialTopic
トピックに公開します。
echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null
次に、kafka-console-consumer.sh
スクリプトを使用してKafkaコンシューマーを作成できます。 ZooKeeperサーバーのホスト名とポート、および引数としてのトピック名が必要です。
次のコマンドは、TutorialTopic
からのメッセージを消費します。 --from-beginning
フラグの使用に注意してください。これにより、コンシューマーが開始される前に公開されたメッセージを使用できます。
~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TutorialTopic --from-beginning
構成の問題がない場合は、端末にHello, World
が表示されます。
OutputHello, World
スクリプトは引き続き実行され、トピックにさらにメッセージが公開されるのを待ちます。 新しいターミナルを開いてプロデューサーを開始し、さらにいくつかのメッセージを公開してください。 コンシューマーの出力でそれらすべてを確認できるはずです。
テストが終了したら、CTRL+C
を押してコンシューマスクリプトを停止します。 インストールをテストしたので、KafkaTのインストールに移りましょう。
ステップ6— KafkaTのインストール(オプション)
KafkaT はAirbnbのツールで、Kafkaクラスターの詳細を表示したり、コマンドラインから特定の管理タスクを実行したりするのが簡単になります。 これはRubygemであるため、使用するにはRubyが必要になります。 また、ruby-devel
と、make
やgcc
などのビルド関連パッケージを使用して、依存する他のgemをビルドできるようにする必要があります。 yum
を使用してそれらをインストールします。
sudo yum install ruby ruby-devel make gcc patch
これで、gemコマンドを使用してKafkaTをインストールできます。
sudo gem install kafkat
KafkaTは、構成ファイルとして.kafkatcfg
を使用して、Kafkaサーバーのインストールディレクトリとログディレクトリを決定します。 また、KafkaTをZooKeeperインスタンスにポイントするエントリが必要です。
.kafkatcfg
という名前の新しいファイルを作成します。
vi ~/.kafkatcfg
次の行を追加して、KafkaサーバーとZookeeperインスタンスに関する必要な情報を指定します。
〜/ .kafkatcfg
{ "kafka_path": "~/kafka", "log_path": "/tmp/kafka-logs", "zk_path": "localhost:2181" }
編集が終了したら、ファイルを保存して閉じます。
これで、KafkaTを使用する準備が整いました。 まず、これを使用してすべてのKafkaパーティションの詳細を表示する方法を次に示します。
kafkat partitions
次の出力が表示されます。
OutputTopic Partition Leader Replicas ISRs TutorialTopic 0 0 [0] [0] __consumer_offsets 0 0 [0] [0] ... ...
TutorialTopic
と、Kafkaがクライアント関連情報を保存するために使用する内部トピックである__consumer_offsets
が表示されます。 __consumer_offsets
で始まる行は無視してかまいません。
KafkaTの詳細については、KafkaTのGitHubリポジトリを参照してください。
ステップ7—マルチノードクラスターのセットアップ(オプション)
より多くのCentOS7マシンを使用してマルチブローカークラスターを作成する場合は、新しいマシンごとにステップ1、ステップ4、およびステップ5を繰り返す必要があります。 さらに、それぞれのserver.properties
ファイルに次の変更を加える必要があります。
broker.id
プロパティの値は、クラスター全体で一意になるように変更する必要があります。 このプロパティは、クラスター内の各サーバーを一意に識別し、その値として任意の文字列を持つことができます。 たとえば、"server1"
、"server2"
などです。zookeeper.connect
プロパティの値は、すべてのノードが同じZooKeeperインスタンスを指すように変更する必要があります。 このプロパティは、Zookeeperインスタンスのアドレスを指定し、<HOSTNAME/IP_ADDRESS>:<PORT>
形式に従います。 たとえば、"203.0.113.0:2181"
、"203.0.113.1:2181"
などです。
クラスタに複数のZooKeeperインスタンスが必要な場合は、各ノードのzookeeper.connect
プロパティの値は、すべてのZooKeeperインスタンスのIPアドレスとポート番号をリストする同一のコンマ区切りの文字列である必要があります。
ステップ8—Kafkaユーザーを制限する
すべてのインストールが完了したので、kafkaユーザーの管理者権限を削除できます。 その前に、root以外のsudoユーザーとしてログアウトして再度ログインしてください。 このチュートリアルを開始したのと同じシェルセッションをまだ実行している場合は、exit
と入力するだけです。
kafkaユーザーをsudoグループから削除します。
sudo gpasswd -d kafka wheel
Kafkaサーバーのセキュリティをさらに向上させるには、passwd
コマンドを使用してkafkaユーザーのパスワードをロックします。 これにより、誰もこのアカウントを使用してサーバーに直接ログインできないようになります。
sudo passwd kafka -l
この時点で、 rootまたはsudo
ユーザーのみが、次のコマンドを入力してkafka
としてログインできます。
sudo su - kafka
将来、ロックを解除する場合は、passwd
を-u
オプションとともに使用します。
sudo passwd kafka -u
これで、kafkaユーザーの管理者権限が正常に制限されました。
結論
これで、CentOSサーバーでApacheKafkaが安全に実行されます。 ほとんどのプログラミング言語で利用可能なKafkaクライアントを使用してKafkaプロデューサーとコンシューマーを作成することにより、プロジェクトでそれを利用できます。 Kafkaの詳細については、Kafkaのドキュメントを参照することもできます。