Ubuntu 18.04にApache Kafkaをインストールする方法

Ubuntu 18.04を使用していませんか？別のバージョンやディストリビューションを選択してください。

著者は、Write for DOnationsプログラムの一環として寄付を受けるために、Free and Open Source Fundを選びました。

紹介

Apache Kafkaは、大量のリアルタイムデータを効率的に処理するために設計された人気のある分散メッセージブローカーです。Kafkaクラスターは、非常にスケーラブルで障害耐性が高いだけでなく、ActiveMQやRabbitMQなどの他のメッセージブローカーと比較して、はるかに高いスループットを持っています。一般的にはパブリッシュ/サブスクライブメッセージングシステムとして使用されていますが、多くの組織が公開されたメッセージの永続的なストレージを提供するため、ログ集約にも利用しています。

パブリッシュ/サブスクライブメッセージングシステムは、1つ以上のプロデューサーがメッセージを公開することを可能にしますが、消費者の数やそれらがメッセージをどのように処理するかを考慮する必要はありません。サブスクライブしたクライアントは、更新や新しいメッセージの作成について自動的に通知されます。このシステムは、クライアントが定期的にポーリングして新しいメッセージが利用可能かどうかを確認するシステムよりも、より効率的でスケーラブルです。

このチュートリアルでは、Ubuntu 18.04にApache Kafka 2.1.1をインストールして使用します。

前提条件

続けるためには、次が必要です:

Ubuntu 18.04 サーバーと sudo 権限を持つ非ルートユーザーが必要です。非ルートユーザーが設定されていない場合は、このガイドに指定された手順に従ってください。
サーバーには少なくとも4GBのRAMが必要です。このRAMの量がないインストールでは、Kafkaサービスが失敗する可能性があり、Java仮想マシン（JVM）が起動時に「Out Of Memory」例外をスローします。
サーバーにOpenJDK 8がインストールされています。このバージョンをインストールするには、特定のバージョンのOpenJDKのインストールに関する指示に従ってください。KafkaはJavaで書かれているため、JVMを必要としますが、その起動シェルスクリプトにはバージョン検出のバグがあり、JVMバージョン8以上では起動に失敗します。

ステップ 1 — Kafka のユーザーを作成する

Kafkaはネットワーク越しのリクエストを処理できるため、専用のユーザーを作成すべきです。これは、Kafkaサーバーが侵害された場合にUbuntuマシンへの被害を最小限に抑えるためです。このステップでは専用のkafkaユーザーを作成しますが、Kafkaの設定が完了した後、このサーバーで他のタスクを実行するために異なる非rootユーザーを作成するべきです。

非ルートのsudoユーザーとしてログインして、kafkaというユーザーをuseraddコマンドを使って作成します:

sudo useradd kafka -m

-mフラグは、ユーザーのためにホームディレクトリが作成されることを保証します。このホームディレクトリ、/home/kafkaは、以下のセクションでコマンドを実行するための作業ディレクトリとして機能します。

passwdを使用してパスワードを設定してください:

sudo passwd kafka

kafka ユーザーを adduser コマンドを使用して sudo グループに追加し、Kafka の依存関係をインストールするために必要な権限を与えます:

sudo adduser kafka sudo

あなたの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

私たちは、アーカイブの内容が別のディレクトリ（例えば、~/kafka/kafka_2.11-2.1.1/）の中にではなく、~/kafka/自体に抽出されることを確実にするために、--strip 1フラグを指定します。

バイナリを正常にダウンロードして抽出したので、トピックの削除を許可するようにKafkaを設定する作業に進むことができます。

ステップ3 — Kafkaサーバーの設定

Kafkaのデフォルトの動作では、メッセージを公開できるカテゴリ、グループ、またはフィード名であるトピックを削除することはできません。これを修正するために、設定ファイルを編集しましょう。

Kafkaの設定オプションはserver.propertiesに指定されています。このファイルをnanoまたはお好みのエディタで開きます:

nano ~/kafka/config/server.properties

Kafkaトピックを削除するための設定を追加しましょう。ファイルの下部に次の内容を追加してください:

delete.topic.enable = true

ファイルを保存して、nanoを終了します。Kafkaを設定したので、次にsystemdユニットファイルを作成して、起動時に実行および有効化することに進むことができます。

ステップ4 — Systemdユニットファイルの作成とKafkaサーバーの起動

このセクションでは、Kafkaサービスのためのsystemdユニットファイルを作成します。これにより、他のLinuxサービスと一貫性を持ってKafkaを起動、停止、再起動するなどの一般的なサービスアクションを実行できるようになります。

Zookeeperは、Kafkaがクラスタの状態と設定を管理するために使用するサービスです。これは多くの分散システムで不可欠なコンポーネントとして一般的に使用されています。詳細については、公式のZookeeperドキュメントをご覧ください。

zookeeperのユニットファイルを作成します:

sudo nano /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 nano /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に依存していることを指定します。これにより、kafkaサービスが起動する際にzookeeperが自動的に開始されることが保証されます。

＜code>[Service]＜/code＞セクションは、systemdがサービスを開始および停止するために＜code>kafka-server-start.sh＜/code＞と＜code>kafka-server-stop.sh＜/code＞シェルファイルを使用することを指定しています。また、Kafkaが異常終了した場合には自動的に再起動されるべきであることも指定されています。

単位が定義されたので、以下のコマンドでKafkaを起動してください:

sudo systemctl start kafka

サーバーが正常に起動したことを確認するために、kafka ユニットのジャーナルログを確認してください:

sudo journalctl -u kafka

次のような出力が表示されるはずです。

Jul 17 18:38:59 kafka-ubuntu systemd[1]: Started kafka.service.

あなたは現在、ポート9092でリスニングしているKafkaサーバーを持っています。

私たちがkafkaサービスを始めたとしても、サーバーを再起動すると自動的には開始されません。サーバーの起動時にkafkaを有効にするには、次のコマンドを実行してください:

sudo systemctl enable kafka

サービスを開始し有効にしたので、インストールを確認しましょう。

ステップ5 — インストールのテスト

Kafkaサーバーが正しく動作していることを確認するために、“Hello World ”メッセージを公開して消費しましょう。Kafkaでメッセージを公開するには次のものが必要です：

プロデューサーは、トピックに関する記録やデータの公開を可能にします。
トピックからメッセージやデータを読み取る消費者。

まず、TutorialTopicというトピックを次のように入力して作成します:

~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic TutorialTopic

コマンドラインから kafka-console-producer.sh スクリプトを使用してプロデューサーを作成できます。これは、Kafkaサーバーのホスト名、ポート、およびトピック名を引数として期待します。

トピックに文字列"Hello, World"を公開するには、次のように入力します:

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が表示されるはずです。

Hello, World

スクリプトは継続して実行され、トピックに公開される追加のメッセージを待機します。新しいターミナルを開いて、いくつかのメッセージを公開するプロデューサーを開始しても構いません。すべてのメッセージが消費者の出力に表示されるはずです。

テストが完了したら、CTRL+C を押して消費者スクリプトを停止してください。インストールのテストが完了したので、次はKafkaTのインストールに進みましょう。

ステップ6 — KafkaTをインストールする（オプション）

KafkaTはAirbnbのツールで、Kafkaクラスタの詳細を表示し、コマンドラインから特定の管理タスクを実行するのを容易にします。それはRubyのgemであるため、使用するにはRubyが必要です。また、依存している他のgemをビルドするためにbuild-essentialパッケージも必要です。これらをaptを使ってインストールしてください:

sudo apt install ruby ruby-dev build-essential

今、gemコマンドを使用してKafkaTをインストールできます:

sudo gem install kafkat

KafkaTは、Kafkaサーバーのインストールおよびログディレクトリを決定するために、.kafkatcfgを構成ファイルとして使用します。また、KafkaTがあなたのZooKeeperインスタンスを指すエントリも含める必要があります。

.kafkatcfgという新しいファイルを作成してください:

nano ~/.kafkatcfg

KafkaサーバーとZookeeperインスタンスに関する必要な情報を指定するために、次の行を追加してください:

{
  "kafka_path": "~/kafka",
  "log_path": "/tmp/kafka-logs",
  "zk_path": "localhost:2181"
}

あなたは今、KafkaTを使用する準備が整いました。まずは、すべてのKafkaパーティションの詳細を表示する方法を紹介します:

kafkat partitions

次の出力が表示されます:

Topic                 Partition   Leader      Replicas        ISRs    
TutorialTopic         0             0         [0]             [0]
__consumer_offsets	  0		          0		      [0]							[0]
...
...

あなたはTutorialTopicや、クライアント関連の情報を保存するためにKafkaによって使用される内部トピック__consumer_offsetsを見ることになります。__consumer_offsetsで始まる行は安全に無視して構いません。

KafkaTについてさらに学ぶには、そのGitHubリポジトリを参照してください。

ステップ7 — マルチノードクラスターの設定（オプション）

Ubuntu 18.04のマシンを追加してマルチブローカークラスターを作成したい場合は、各新しいマシンでステップ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ユーザーの管理者権限を削除できます。その前に、他の非ルートのsudoユーザーとしてログアウトし、再ログインしてください。チュートリアルを開始した同じシェルセッションがまだ実行中の場合は、単にexitと入力してください。

kafka ユーザーを sudo グループから削除します:

sudo deluser kafka sudo

Kafkaサーバーのセキュリティをさらに向上させるために、kafkaユーザーのパスワードをpasswdコマンドを使用してロックします。これにより、このアカウントを使用してサーバーに直接ログインできる人がいなくなります:

sudo passwd kafka -l

この時点では、ルートユーザーまたはsudoユーザーのみが次のコマンドを入力してkafkaとしてログインできます。

sudo su - kafka

将来的にこれを解除したい場合は、passwdコマンドを-uオプションと共に使用してください:

sudo passwd kafka -u

あなたは現在、kafkaユーザーの管理者権限を正常に制限しました。

結論

あなたは現在、Ubuntuサーバー上で安全にApache Kafkaを実行しています。ほとんどのプログラミング言語で利用可能なKafkaクライアントを使って、Kafkaプロデューサーとコンシューマーを作成することで、プロジェクトで利用することができます。Kafkaについてさらに学ぶには、そのドキュメントも参照できます。