ApsaraMQ for Kafka:VPC 内で ApsaraMQ for Kafka を Logstash の入力として使用

ステップ 1：エンドポイントの取得

Logstash は VPC エンドポイントを介して ApsaraMQ for Kafka に接続します。ApsaraMQ for Kafka は、2 種類の VPC エンドポイントを提供します：

詳細については、「エンドポイント間の比較」をご参照ください。

エンドポイントを取得するには：

1. ApsaraMQ for Kafka コンソールにログインします。

2. [概要] ページの [リソース配布] セクションで、インスタンスがデプロイされているリージョンを選択します。

3. [インスタンス] ページで、ターゲットインスタンスの名前をクリックします。

4. [インスタンス詳細] ページで、[エンドポイント情報] セクションのエンドポイントを見つけます。SASL エンドポイントを使用する場合は、[設定情報] セクションの [ユーザー名] と [パスワード] の値をメモしておきます。

ステップ 2：トピックの作成

Logstash が消費するメッセージを保持するためのトピックを作成します。

1. ApsaraMQ for Kafka コンソールにログインします。

2. [概要] ページの [リソース配布] セクションで、インスタンスがデプロイされているリージョンを選択します。

3. [インスタンス] ページで、ターゲットインスタンスの名前をクリックします。

4. 左側のナビゲーションウィンドウで、[Topics] をクリックします。

5. [Topics] ページで、[トピックの作成] をクリックします。

6. [トピックの作成] パネルで、次のパラメーターを設定し、[OK] をクリックします。

トピックが作成されると、[Topics] ページに表示されます。

ステップ 3：テストメッセージの送信

テストメッセージを送信して、トピックが消費可能な状態であることを確認します。

1. ApsaraMQ for Kafka コンソールにログインします。

2. ターゲットインスタンスに移動し、左側のナビゲーションウィンドウで [Topics] をクリックします。

3. トピック名をクリックし、[トピック詳細] ページの右上隅にある [メッセージの送信] をクリックします。

4. [メッセージの送受信を開始] パネルで、[送信方法] ([コンソール]、[Docker]、または [SDK]) を選択してテストメッセージを送信します。 メッセージを特定のパーティションに送信する場合は、パーティション ID を指定できます。 パーティション ID のクエリ方法については、「パーティションステータスの表示」をご参照ください。

ステップ 4：コンシューマーグループの作成

Logstash が消費オフセットを追跡するためのコンシューマーグループを作成します。

1. ApsaraMQ for Kafka コンソールにログインします。

2. ターゲットインスタンスに移動し、左側のナビゲーションウィンドウで [Groups] をクリックします。

3. [Groups] ページで、[グループの作成] をクリックします。

4. [グループの作成] パネルで、[グループ ID] と [説明] を入力し、必要に応じてタグをアタッチして、[OK] をクリックします。

コンシューマーグループが作成されると、[Groups] ページに表示されます。

ステップ 5：Logstash の設定と実行

1. Logstash のインストールディレクトリに移動します。

2. bin ディレクトリに input.conf という名前の設定ファイルを作成します：

input {
  kafka {
    bootstrap_servers => "<your-kafka-endpoint>"       # VPC エンドポイント (例：alikafka-pre-cn-zv****-1-vpc.alikafka.aliyuncs.com:9092)
    group_id          => "<your-consumer-group-id>"     # ステップ 4 で作成したコンシューマーグループ
    topics            => ["<your-topic-name>"]          # ステップ 2 で作成したトピック
    consumer_threads  => 12                             # トピックのパーティション数に合わせてください
    auto_offset_reset => "earliest"                     # 最も古いメッセージから開始
  }
}
output {
  stdout { codec => rubydebug }
}

1. Logstash を実行します：

./logstash -f input.conf

メッセージ消費の確認

Logstash が起動すると、消費されたメッセージが Ruby デバッグ形式で標準出力 (stdout) に出力されます：

関連情報

• Kafka 入力プラグイン -- Logstash の Kafka 入力プラグインの完全なパラメーターリファレンス。

• エンドポイント間の比較 -- ApsaraMQ for Kafkaのエンドポイントタイプの違い。