Logstash を使用して Kafka データを AnalyticDB for MySQL に書き込む - AnalyticDB

データがすでに Apache Kafka パイプラインを通過している場合、Logstash を使用して AnalyticDB for MySQL に直接ルーティングできます。AnalyticDB for MySQL は MySQL と完全な互換性があるため、標準の Logstash JDBC 出力プラグインは、追加のドライバーやアダプターなしで接続できます。

Logstash によるデータ処理の仕組み

Logstash は、以下の 3 つの連続したプラグインステージを介してデータを移動させます：

ステージ	プラグインタイプ	ロール
収集	入力	ソース (Kafka、ファイル、HTTP など) からイベントを読み取ります。
変換	フィルター	イベントの解析、エンリッチ、再形成 (Grok 解析、IP ジオロケーション、PII の匿名化) を行います。
書き込み	出力	変換されたイベントを送信先 (JDBC 経由の AnalyticDB for MySQL) に送信します。

前提条件

開始する前に、以下が準備できていることを確認してください：

Logstash 1.5 以降 (バージョン 1.5 で Apache Kafka の組み込み統合が導入されました)
インポートするデータを含むトピックが少なくとも 1 つある Apache Kafka クラスター
AnalyticDB for MySQL のクラスターエンドポイント、データベース名、ユーザー名、パスワード
Logstash のクラスパス上にある MySQL JDBC ドライバー (com.mysql.jdbc.Driver)

Kafka データの AnalyticDB for MySQL へのインポート

ステップ 1：プラグインのインストールと更新

Apache Kafka サーバーのルートディレクトリで、以下を実行します：

bin/plugin install
bin/plugin update

利用可能な Logstash プラグインの完全なリストについては、「GitHub の Logstash プラグインリポジトリ」をご参照ください。

ステップ 2：パイプラインの設定

Logstash 設定ファイル (例：config/kafka-to-adb.conf) を作成し、入力ブロックと出力ブロックを設定します。

入力設定

以下の例では、Kafka コンシューマーを設定します：

input {
    kafka {
        zk_connect => "localhost:2181"
        group_id => "Logstash"
        topic_id => "test"
        codec => plain
        reset_beginning => false # boolean (任意)、デフォルト：false
        consumer_threads => 5 # number (任意)、デフォルト：1
        decorate_events => true # boolean (任意)、デフォルト：false
    }
}

入力パラメーター

パラメーター	型	必須	デフォルト	説明
`zk_connect`	文字列	はい	—	ZooKeeper 接続文字列。例：`localhost:2181`。
`group_id`	文字列	はい	—	コンシューマーグループ ID。異なるコンシューマーグループ間での消費は互いに分離されます。
`topic_id`	文字列	はい	—	サブスクライブして消費する Kafka トピック。
`reset_beginning`	boolean	いいえ	`false`	Logstash 起動時のオフセット位置：`false` は最後にコミットされたオフセットから再開します (以前のオフセットがない場合は最小オフセットから)。`true` は最小オフセットから開始し、最後のメッセージを消費した後にフォローモード (`tail -F`) に切り替わります。
`consumer_threads`	number	いいえ	`1`	並列コンシューマースレッドの数。
`decorate_events`	boolean	いいえ	`false`	`true` の場合、各イベントにメタデータ (メッセージサイズ、トピックソース、コンシューマーグループ) をアタッチします。
`rebalance_max_retries`	number	いいえ	—	コンシューマーグループのリバランス中に、ZooKeeper にパーティションオーナーノードを登録する際のリトライ回数。
`consumer_timeout_ms`	number	いいえ	—	メッセージ受信のタイムアウト期間。デフォルト値でご利用の環境に問題が発生する場合にのみ、この値を変更してください。

完全なパラメーターリファレンスについては、「GitHub の logstash-kafka README」をご参照ください。

説明

同じトピックからメッセージを並行して消費するには、トピックを複数のパーティションに分割し、各コンシューマーに同じ group_id と topic_id を割り当てます。これにより、メッセージが順番に消費されることが保証されます。

出力設定

以下の例では、JDBC 出力プラグインを使用して AnalyticDB for MySQL にイベントを書き込みます：

output {
    jdbc {
        driver_class => "com.mysql.jdbc.Driver"
        connection_string => "jdbc:mysql://HOSTNAME/DATABASE?user=USER&password=PASSWORD"
        statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, ?, ?)", "host", "@timestamp", "message" ]
    }
}

connection_string 内のプレースホルダーを置き換えます：

プレースホルダー	説明
`HOSTNAME`	AnalyticDB for MySQL クラスターのエンドポイント
`DATABASE`	ターゲットデータベース名
`USER`	データベースのユーザー名
`PASSWORD`	データベースのパスワード

出力パラメーター

パラメーター	型	必須	説明
`driver_class`	文字列	はい	JDBC ドライバーのクラス名。AnalyticDB for MySQL には `com.mysql.jdbc.Driver` を使用します。
`connection_string`	文字列	はい	AnalyticDB for MySQL クラスターの JDBC 接続 URL。
`statement`	配列	はい	最初の要素が `?` プレースホルダー付きの INSERT 文である配列。その後に、各プレースホルダーに順番にマッピングされる Logstash フィールド名が続きます。

完全なパラメーターリファレンスについては、「GitHub の logstash-kafka README」をご参照ください。

ステップ 3：パイプラインの開始

Logstash のインストールディレクトリで、以下を実行します：

bin/Logstash -f config/kafka-to-adb.conf

Logstash は、設定された Kafka トピックからメッセージの消費を開始し、AnalyticDB for MySQL に書き込みます。