ウェブサイト検索

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

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

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

紹介

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

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

このチュートリアルでは、Ubuntu 20.04にApache Kafka 2.8.2をインストールし、設定します。

前提条件

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

  • 少なくとも4GBのRAMを搭載したUbuntu 20.04サーバーと、sudo権限を持つ非rootユーザーが必要です。非rootユーザーが設定されていない場合は、初期サーバーセットアップガイドに従ってこれを設定できます。4GB未満のRAMを搭載したインストールでは、Kafkaサービスが失敗する可能性があります。
  • OpenJDK 11がサーバーにインストールされています。このバージョンをインストールするには、Ubuntu 20.04でAPTを使用してJavaをインストールする方法に関する当社のチュートリアルに従ってください。KafkaはJavaで書かれているため、JVMが必要です。

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

Kafkaはネットワーク越しにリクエストを処理できるため、最初のステップはサービス用の専用ユーザーを作成することです。これにより、誰かがKafkaサーバーを侵害した場合にUbuntuマシンへの損害を最小限に抑えることができます。このステップでは、専用のkafkaユーザーを作成します。

非ルートのsudoユーザーとしてサーバーにログインし、kafkaというユーザーを作成してください:

  1. sudo adduser kafka

プロンプトに従ってパスワードを設定し、kafkaユーザーを作成してください。

次に、adduserコマンドを使用してkafkaユーザーをsudoグループに追加します。Kafkaの依存関係をインストールするにはこれらの権限が必要です:

  1. sudo adduser kafka sudo

あなたのkafkaユーザーは準備が整いました。suを使用してkafkaアカウントにログインしてください:

  1. su -l kafka

Kafka専用のユーザーを作成したので、Kafkaのバイナリをダウンロードして展開する準備が整いました。

ステップ2 — Kafkaバイナリのダウンロードと抽出

このステップでは、Kafkaのバイナリをダウンロードして展開し、kafkaユーザーのホームディレクトリ内の専用フォルダーに配置します。

まず、/home/kafka に Downloads というディレクトリを作成して、ダウンロードを保存します:

  1. mkdir ~/Downloads

Kafkaのバイナリをダウンロードするには、curlを使用してください。

  1. curl "https://downloads.apache.org/kafka/2.8.2/kafka_2.13-2.8.2.tgz" -o ~/Downloads/kafka.tgz

「kafka」という名前のディレクトリを作成し、このディレクトリに移動します。このディレクトリをKafkaのインストールの基準ディレクトリとして使用します:

  1. mkdir ~/kafka && cd ~/kafka

ダウンロードしたアーカイブをtarコマンドを使用して展開します:

  1. tar -xvzf ~/Downloads/kafka.tgz --strip 1

アーカイブの内容を別のディレクトリ（例： ~/kafka/kafka_2.13-2.8.2/ のような）ではなく、直接 ~/kafka/ に抽出するために、--strip 1 フラグを指定します。

バイナリを正常にダウンロードして抽出したので、Kafkaサーバーの設定を開始できます。

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

Kafkaのトピックは、メッセージが公開されるカテゴリ、グループ、またはフィード名です。しかし、Kafkaのデフォルト設定ではトピックを削除することはできません。これを変更するには、設定ファイルを編集する必要があります。これはこのステップで行います。

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

  1. nano ~/kafka/config/server.properties

最初に、Kafkaトピックを削除できる設定を追加します。ファイルの一番下に次の行を追加してください:

delete.topic.enable = true

次に、log.dirsプロパティを変更して、Kafkaログが保存されるディレクトリを変更します。log.dirsプロパティを見つけて、既存のルートをハイライトされたルートに置き換えてください:

log.dirs=/home/kafka/logs

ファイルを保存して閉じてください。

Kafkaを設定したので、Kafkaサーバーを起動時に実行および有効化するためのsystemdユニットファイルを作成できます。

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

このセクションでは、Kafkaサービスのためのsystemdユニットファイルを作成します。これらのファイルは、他のLinuxサービスと一貫した方法でKafkaを起動、停止、再起動するなどの一般的なサービスアクションを実行するのに役立ちます。

KafkaはZookeeperを使用してクラスタの状態と構成を管理します。これは多くの分散システムで使用されており、公式のZookeeperドキュメントでこのツールについてもっと読むことができます。これらのユニットファイルを使用してZookeperをサービスとして使用します。

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

  1. 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サービスファイルを作成します:

  1. 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 が自動的に起動することが保証されます。

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

ファイルを保存して閉じてください。

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

  1. sudo systemctl start kafka

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

  1. sudo systemctl status kafka

あなたはこのような出力を受け取ります:

● kafka.service
     Loaded: loaded (/etc/systemd/system/kafka.service; disabled; vendor preset>
     Active: active (running) since Wed 2023-02-01 23:44:12 UTC; 4s ago
   Main PID: 17770 (sh)
      Tasks: 69 (limit: 4677)
     Memory: 321.9M
     CGroup: /system.slice/kafka.service
             ├─17770 /bin/sh -c /home/kafka/kafka/bin/kafka-server-start.sh /ho>
             └─17793 java -Xmx1G -Xms1G -server -XX:+UseG1GC -XX:MaxGCPauseMill>

現在、9092番ポートでリッスンしているKafkaサーバーがあります。これはKafkaサーバーが使用するデフォルトのポートです。

あなたはkafkaサービスを始めました。しかし、サーバーを再起動すると、Kafkaは自動的に再起動しません。サーバーのブート時にkafkaサービスを有効にするには、次のコマンドを実行してください:

  1. sudo systemctl enable zookeeper

シンボリックリンクが作成されたという応答が得られます:

Created symlink /etc/systemd/system/multi-user.target.wants/zookeeper.service → /etc/systemd/system/zookeeper.service.

次のコマンドを実行してください:

  1. sudo systemctl enable kafka

シンボリックリンクが作成されたという応答が得られます:

Created symlink /etc/systemd/system/multi-user.target.wants/kafka.service → /etc/systemd/system/kafka.service.

このステップでは、kafkaおよびzookeeperサービスを開始して有効にしました。次のステップでは、Kafkaのインストールを確認します。

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

このステップでは、Kafkaのインストールをテストします。Kafkaサーバーが期待通りに動作していることを確認するために、Hello World メッセージを公開し、消費します。

Kafkaにメッセージを公開するには:

  • トピックに関するレコードやデータの出版を可能にするプロデューサー。
  • トピックからメッセージやデータを読み取る消費者。

まず、TutorialTopicという名前のトピックを作成します。

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

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

トピックが作成されたという応答を受け取るでしょう。

Created topic TutorialTopic.

次に、"Hello, World"という文字列をTutorialTopicトピックに公開します:

  1. 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フラグを使用することに注意してください:

  1. ~/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TutorialTopic --from-beginning

設定に問題がなければ、ターミナルにHello, Worldという応答が表示されます:

Hello, World

スクリプトは実行を続け、公開するためのメッセージを待機します。これをテストするには、新しいターミナルウィンドウを開き、サーバーにログインしてください。kafka ユーザーとしてログインすることを忘れないでください:

  1. su -l kafka

この新しいターミナルで、2番目のメッセージを公開するためにプロデューサーを開始します:

  1. echo "Hello World from Sammy at DigitalOcean!" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TutorialTopic > /dev/null

このメッセージは、あなたの元の端末の消費者の出力に読み込まれます:

Hello, World
Hello World from Sammy at DigitalOcean!

テストが終わったら、元のターミナルで消費者スクリプトを停止するには CTRL+C を押してください。

あなたは今、Ubuntu 20.04にKafkaサーバーをインストールして設定しました。次のステップでは、Kafkaサーバーのセキュリティを強化するためのいくつかの簡単な作業を行います。

ステップ6 — Kafkaサーバーの強化

インストールが完了したので、kafka ユーザーの管理権限を削除し、Kafka サーバーを強化できます。

その前に、ログアウトして他の非ルートの sudo ユーザーとして再ログインしてください。このチュートリアルを始めたときと同じシェルセッションを実行している場合は、exit と入力してください。

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

  1. sudo deluser kafka sudo

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

  1. sudo passwd kafka -l

-lフラグはユーザーのパスワードを変更するコマンド（passwd）をロックします。

この時点で、kafkaとしてログインできるのは、rootまたはsudoユーザーのみです。次のコマンドを使用します:

  1. sudo su - kafka

将来的に、パスワードを変更する能力をアンロックしたい場合は、passwdコマンドを-uオプションと共に使用してください:

  1. sudo passwd kafka -u

あなたは現在、kafkaユーザーの管理権限を正常に制限しました。Kafkaの使用を開始する準備が整いました。次のステップをオプションで実行すると、KafkaTをシステムに追加できます。

ステップ7 — KafkaTのインストール（オプション）

KafkaTは、Kafkaクラスターに関する詳細を表示したり、コマンドラインから特定の管理タスクを実行したりする能力を向上させるために開発されました。これはRubyのgemであるため、使用するにはRubyが必要です。また、KafkaTが依存する他のgemをビルドするために、build-essentialパッケージも必要になります。

Rubyとbuild-essentialパッケージをaptを使ってインストールします。

  1. sudo apt install ruby ruby-dev build-essential

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

  1. sudo CFLAGS=-Wno-error=format-overflow gem install kafkat

Wno-error=format-overflow コンパイルフラグは、kafkat のインストールプロセス中に Zookeeper の警告とエラーを抑制するために必要です。

インストールが完了すると、完了したという応答が届きます。

...
Done installing documentation for json, colored, retryable, highline, trollop, zookeeper, zk, kafkat after 3 seconds
8 gems installed

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

.kafkatcfgという名前の新しいファイルを作成します。

  1. nano ~/.kafkatcfg

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

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

ファイルを保存して閉じてください。これでKafkaTを使用する準備が整いました。

すべてのKafkaパーティションの詳細を表示するには、このコマンドを実行してみてください:

  1. kafkat partitions

次の出力を受け取ります:

[DEPRECATION] The trollop gem has been renamed to optimist and will no longer be supported. Please switch to optimist as soon as possible.
/var/lib/gems/2.7.0/gems/json-1.8.6/lib/json/common.rb:155: warning: Using the last argument as keyword parameters is deprecated
...
Topic                 Partition   Leader      Replicas        ISRs
TutorialTopic         0             0         [0]             [0]
__consumer_offsets	  0		          0		      [0]							[0]
...
...

出力にはTutorialTopicと__consumer_offsetsが含まれます。これは、クライアント関連の情報を保存するためにKafkaが使用する内部トピックです。__consumer_offsetsで始まる行は安全に無視できます。

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

結論

あなたは現在、UbuntuサーバーでApache Kafkaを安全に実行しています。Kafkaクライアントを使用して、お気に入りのプログラミング言語にKafkaを統合できます。

Kafkaについてもっと学ぶには、そのドキュメントも参照できます。

関連記事: