データが既に Apache Kafka パイプラインを経由してフローしている場合、Logstash を使用して、AnalyticDB for MySQL Data Warehouse Edition (V3.0) に直接ルーティングできます。AnalyticDB for MySQL は MySQL と完全互換であるため、標準の Logstash JDBC 出力プラグインを追加のドライバーまたはアダプターなしで接続できます。
Logstash によるデータ処理の仕組み
Logstash は、3 つの順次実行されるプラグインステージを経てデータを移動させます：
| ステージ | プラグイン種別 | 役割 |
|---|---|---|
| 収集 | 入力 | ソース（Kafka、ファイル、HTTP など）からイベントを読み取ります |
| 変換 | フィルター（Filter） | イベントの解析、エンリッチメント、再構成（Grok 解析、IP ジオロケーション、PII の匿名化など）を行います |
| 書き込み | 出力（Output） | 変換済みのイベントを送信先（JDBC 経由の AnalyticDB for MySQL）に送信します |
前提条件
開始する前に、以下の環境が整っていることを確認してください。
Logstash 1.5 以降（バージョン 1.5 で Apache Kafka 統合機能が標準搭載されました）
インポート対象のデータを含む Topic を 1 つ以上持つ Apache Kafka クラスター
AnalyticDB for MySQL Data Warehouse Edition (V3.0) クラスターのエンドポイント、データベース名、ユーザー名、およびパスワード
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 # 数値（任意）、デフォルト：1
decorate_events => true # boolean（任意）、デフォルト：false
}
}入力パラメーター
| パラメーター | 型 | 必須 | デフォルト | 説明 |
|---|---|---|---|---|
zk_connect | 文字列 | はい | — | ZooKeeper 接続文字列（例：localhost:2181） |
group_id | 文字列 | はい | — | コンシューマーグループ ID。異なるコンシューマーグループ間の消費は相互に分離されます。 |
topic_id | 文字列 | はい | — | サブスクライブおよび消費対象となる Kafka Topic |
reset_beginning | boolean | いいえ | false | Logstash 起動時のオフセット位置：false の場合、直近のコミット済みオフセット（または未コミットの場合は最小オフセット）から再開します。true の場合、最小オフセットから開始し、最後のメッセージを消費後に「フォロー」モード（tail -F）に切り替わります。 |
consumer_threads | 数値 | いいえ | 1 | 並列実行するコンシューマースレッド数 |
decorate_events | boolean | いいえ | false | true の場合、各イベントにメタデータ（メッセージサイズ、Topic ソース、コンシューマーグループ）をアタッチします。 |
rebalance_max_retries | 数値 | いいえ | — | コンシューマーグループのリバランス時に、ZooKeeper へパーティション所有者ノードを登録する際の最大リトライ回数 |
consumer_timeout_ms | 数値 | いいえ | — | メッセージ受信のタイムアウト期間。デフォルト値がご利用の環境で問題を引き起こす場合にのみ変更してください。 |
パラメーターの詳細なリファレンスについては、「GitHub 上の logstash-kafka README」をご参照ください。
同一 Topic からの並列メッセージ消費を行うには、Topic を複数のパーティションに分割し、各コンシューマーに同一の 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.confLogstash が構成済みの Kafka Topic からメッセージを消費し、AnalyticDB for MySQL へ書き込むようになります。