Rocky LinuxにApache Kafkaをインストールする方法

Rocky LinuxにApache Kafkaをインストールする方法

このページで

1. 前提条件
2. Java OpenJDKのインストール
3. Apache Kafkaをダウンロードしています
4. SystemdサービスとしてKafkaを実行する
5. Kafkaコンソールプロデューサーとコンシューマーを使用した基本操作
6. Kafka Connectプラグインを使用したストリーミングデータ
7. 結論

Apache Kafkaは、リアルタイムでストリーミングデータを処理するための分散データストアです。これはApache Software Foundationによって開発され、JavaとScalaで書かれています。Apache Kafkaは、データストリームに適応するリアルタイムのストリーミングデータパイプラインおよびアプリケーションを構築するために使用され、特にエンタープライズグレードのアプリケーションやミッションクリティカルなアプリケーションに適しています。これは、数千の企業によって使用されている最も人気のあるデータストリームプラットフォームの1つであり、高性能なデータパイプライン、ストリーミング分析、およびデータ統合に利用されています。

Apache Kafkaは、メッセージング、ストレージ、ストリーム処理を1つの場所に統合しています。ユーザーにリアルタイムでデータを収集、処理、およびストリーミングするための高性能で強力なデータストリームを設定することを可能にします。これは、数十億のストリーミングイベントを処理する能力を持つ現代の分散アプリケーションで使用されています。

このチュートリアルでは、Rocky LinuxサーバーにApache Kafkaをインストールし、Kafkaプラグインを介してストリーミングデータのメッセージブローカーとしての基本的な使用方法を学びます。

前提条件

このチュートリアルに従うには、以下の要件が必要です:

• Rocky Linuxサーバー - Rocky Linux v8またはv9を使用できます。
• sudoのルート権限を持つ非ルートユーザー。

Java OpenJDKのインストール

Apache KafkaはJavaベースのアプリケーションです。Kafkaをインストールするには、まずシステムにJavaをインストールする必要があります。本記事執筆時点で、最新のApache Kafkaのバージョンは少なくともJava OpenJDK v11が必要です。

最初のステップでは、公式のRocky LinuxリポジトリからJava OpenJDK 11をインストールします。

以下のdnfコマンドを実行して、Rocky LinuxシステムにJava OpenJDK 11をインストールしてください。

sudo dnf install java-11-openjdk

インストールを確認するように求められたら、y を入力し、ENTER を押して続行します。

Javaがインストールされたら、以下のコマンドを使用してJavaのバージョンを確認してください。あなたのRocky LinuxシステムにはJava OpenJDK 11がインストールされていることが分かります。

java version

Javaがインストールされたので、次にApache Kafkaのインストールを始めます。

Apache Kafkaをダウンロードしています

Apache Kafkaは、Linux/Unixを含む複数のオペレーティングシステム用の複数のバイナリパッケージを提供しています。このステップでは、Kafka用の新しい専用システムユーザーを作成し、Kafkaのバイナリパッケージをダウンロードし、Apache Kafkaのインストールを構成します。

以下のコマンドを実行して、'kafka'という名前の新しいシステムユーザーを作成します。これにより、デフォルトのホームディレクトリ'/opt/kafka'を持つKafka用の新しいシステムユーザーが作成され、これはKafkaのインストールディレクトリとして使用されます。

sudo useradd -r -d /opt/kafka -s /usr/sbin/nologin kafka

次に、作業ディレクトリを'/opt'に移動します。それから、以下のcurlコマンドを使ってApache Kafkaのバイナリパッケージをダウンロードします。これで'kafka.tar.gz'というファイルが表示されるでしょう。

cd /opt
sudo curl -fsSLo kafka.tgz https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz

tar コマンドを使ってファイル 'kafka.tar.gz' を抽出し、抽出したディレクトリを '/opt/kafka' に名前変更してください。

tar -xzf kafka.tgz
sudo mv kafka_2.12-3.3.1 /opt/kafka

次に、以下のchmodコマンドを使用して、'/opt/kafka'ディレクトリの所有権をkafkaユーザーに変更します。

sudo chown -R kafka:kafka /opt/kafka

その後、Apache Kafka用の新しいログディレクトリを作成します。次に、nanoエディタを使用してデフォルトの構成「server.properties」を編集します。

sudo -u kafka mkdir -p /opt/kafka/logs
sudo -u kafka nano /opt/kafka/config/server.properties

KafkaのログディレクトリはApache Kafkaのログを保存するために使用され、kafkaの設定ファイルserver.propertiesでログディレクトリを定義する必要があります。

「log.dirs」オプションのコメントを外し、その値を「/opt/kafka/logs」に変更します。

logs configuration for Apache Kafka
log.dirs=/opt/kafka/logs

作業が完了したら、ファイルを保存してエディタを終了してください。

あなたは現在、Apache Kafkaの基本的なインストールと設定を完了しました。次に、Apache Kafkaをsystemdサービスとして設定して実行します。

SystemdサービスとしてKafkaを実行する

Apache Kafkaパッケージには、サービスを中央集権化し、Kafkaコントローラーの選挙、トピックの設定、およびApache KafkaクラスターのACL（アクセス制御リスト）を維持するために使用される別のアプリケーションZookeeperが含まれています。

Apache Kafkaを実行するには、まずシステムでZookeeperを実行する必要があります。このステップでは、ZookeeperとApache Kafkaのための新しいsystemdサービスファイルを作成します。両方のサービスは同じユーザー「kafka」の下で実行されます。

以下のコマンドを使用して、Zookeeperのための新しいサービスファイルを作成します '/etc/systemd/system/zookeeper.service'.

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=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

作業が完了したら、ファイルを保存してエディタを終了してください。

次に、以下のコマンドを使用してApache Kafkaの新しいサービスファイル「/etc/systemd/system/kafka.service」を作成します。

sudo nano /etc/systemd/system/kafka.service

次の設定をファイルに追加してください。「[Unit]」セクションで、Kafkaサービスはまず「zookeeper.service」が実行されている必要があり、「zookeeper.service」の後は常に実行されていることがわかります。

[Unit]
Requires=zookeeper.service
After=zookeeper.service
[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties > /opt/kafka/logs/start-kafka.log 2>&1'
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal
[Install]
WantedBy=multi-user.target

作業が完了したら、ファイルを保存してエディタを終了してください。

次に、次のsystemctlコマンドを実行してsystemdマネージャを再読み込みし、新しいサービスを適用します。

sudo systemctl daemon-reload

次のコマンドを使用して、ZooKeeperとKafkaのサービスを開始してください。

sudo systemctl start zookeeper
sudo systemctl start kafka

以下のsystemctlコマンドを使用して、KafkaとZookeeperのサービスをシステム起動時に自動的に実行されるように設定します。

sudo systemctl enable zookeeper
sudo systemctl enable kafka

最後に、以下のコマンドを使用して飼育員とKafkaサービスの両方を確認してください。

sudo systemctl status zookeeper
sudo systemctl status kafka

下記の出力では、ゾookeeperサービスの現在のステータスが実行中であり、また有効になっていることがわかります。

以下はkafkaサービスの状態で、現在稼働中であり、サービスも有効です。

Apache Kafkaのインストールが完了し、現在稼働しているので、次にメッセージブローカーとしてのApache Kafkaの基本的な使い方を学びます。また、Kafkaプラグインを使用してリアルタイムでデータをストリーミングする方法も学びます。

Kafkaコンソールプロデューサーとコンシューマーを使用した基本操作

始める前に、この例で使用されるすべてのコマンドは、'/opt/kafka/bin'ディレクトリで入手可能なKafkaパッケージによって提供されます。

このステップでは、Kafkaトピックを作成して一覧表示する方法、プロデューサーを起動してデータを挿入する方法、消費者スクリプトを介してデータをストリーミングする方法、最後にKafkaトピックを削除して環境をクリーンアップする方法を学びます。

以下のコマンドを実行して、新しいKafkaトピックを作成してください。スクリプト「kafka-topics.sh」を使用して、名前「TestTopic」の新しいトピックを1つのレプリケーションとパーティションで作成します。

sudo -u kafka /opt/kafka/bin/kafka-topics.sh \
--create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic TestTopic

次に、Kafka上のトピックのリストを確認するために、以下のコマンドを実行してください。あなたのKafkaサーバーにTestTopicが作成されているはずです。

sudo -u kafka /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

次に、メッセージを生成するには、スクリプト 'kafka-console-producer.sh' を使用し、処理されるデータを挿入します。

次のコマンドを実行してKafkaコンソールプロデューサーを開始し、トピックをTestTopicに指定してください。

sudo -u kafka /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TestTopic

Kafkaコンソールプロデューサーを取得したら、処理されるメッセージを入力してください。

次に、新しいターミナルセッションを開き、サーバーにログインします。それから、スクリプト「kafka-console-consumer.sh」を使ってKafkaコンソールコンシューマーを開きます。

次のコマンドを実行して、Kafkaコンソールコンシューマを開始し、トピックを 'TestTopic' に指定してください。

sudo -u kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic TestTopic --from-beginning

下のスクリーンショットでは、Kafka Console ProducerからのすべてのメッセージがConsumer Consoleに処理されているのを見ることができます。また、Console Producerで他のメッセージを入力すると、そのメッセージが自動的に処理され、Console Consumerの画面に表示されます。

今、Ctrl+cを押してKafkaコンソールプロデューサーとKafkaコンソールコンシューマーの両方を終了してください。

Kafka環境を整理するために、次のコマンドを使用してTestTopicを削除することができます。

sudo -u kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic TestTopic

Kafka Connectプラグインを使用したストリーミングデータ

Apache Kafkaは、複数のソースからデータをストリーミングするために使用できる複数のプラグインを提供します。追加のKafkaライブラリプラグインは、デフォルトで'/opt/kafka/libs'ディレクトリに利用可能であり、設定ファイル'/opt/kafka/config/connect-standalone.properties'を介してkafkaプラグインを有効にする必要があります。この場合、Kafkaスタンドアロンモードのためです。

次のコマンドを実行して、Kafkaの設定ファイル「/opt/kafka/config/connect-standalone.properties」を編集します。

sudo -u kafka nano /opt/kafka/config/connect-standalone.properties

'plugin.path' の行のコメントを外し、値をプラグインのライブラリディレクトリ '/opt/kakfa/libs' に変更してください。

plugin.path=/opt/kafka/libs

作業が完了したら、ファイルを保存してエディタを終了してください。

次に、以下のコマンドを実行して、Kafkaストリームのデータソースとして使用される新しいファイル'/opt/kafka/test.txt'を作成します。

sudo -u kafka echo -e "Test message from file\nTest using Kafka connect from file" > /opt/kafka/test.txt

次に、設定ファイル「connect-file-source.properties」および「connect-file-sink.properties」を使用して、スタンドアロンモードでKafkaコンシューマを開始するために次のコマンドを実行します。

このコマンドと設定は、あなたが作成したソースファイル「test.txt」を使用したKafkaデータストリームのデフォルト例です。この例では、「connect-test」という新しいトピックも自動的に作成され、Kafkaコンソールコンシューマーを介してアクセスできます。

cd /opt/kafka
sudo -u kafka /opt/kafka/bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

さて、別のターミナルセッションを開いて、次のコマンドを実行してKafkaコンソールコンシューマーを起動します。また、トピックを「connect-test」と指定してください。すると、「test.txt」ファイルからのメッセージが表示されます。

sudo -u kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning

これで、ファイル「test.txt」を更新すると、新しいメッセージが自動的に処理され、Kafkaコンソールコンシューマーでストリーミングされるようになります。

以下のコマンドを実行して、ファイル'test.txt'を新しいメッセージで更新してください。

sudo -u kafka echo "Another test message from file test.txt" >> test.txt

以下の出力では、ファイル「test.txt」に変更があるときに、新しいメッセージがKafkaによって自動的に処理される様子を確認できます。これで、ファイルを通じてメッセージをストリーミングするためのKafkaコネクトプラグインの基本的な使用法を完了しました。

結論

このガイドを通じて、Rocky Linux システムに Apache Kafka をインストールする方法を学びました。また、メッセージを生成および処理するための Kafka Producer Console の基本的な使い方や、メッセージを受信するための Kafka Consumer についても学びました。最後に、Kafka プラグインを有効にし、ファイルからリアルタイムでメッセージをストリーミングするための Kafka Connect プラグインの使用方法についても学びました。