Logstash を使用してリアルタイムのログデータを収集し、AnalyticDB for MySQL に書き込む方法 - AnalyticDB

Logstash は、複数のソースからデータを取り込み、変換して、ストレージの宛先に送信する、オープンソースのサーバーサイドデータ処理パイプラインです。AnalyticDB for MySQL は、ネイティブの Java Database Connectivity (JDBC) アクセスをサポートしているため、Logstash 出力プラグインを使用して、ログデータを AnalyticDB for MySQL に直接書き込むことができます。

コミュニティプラグインの logstash-output-jdbc は、1 レコードずつ書き込むため、データ量が多い場合にスループットが低くなり、CPU オーバーヘッドが高くなります。logstash-output-analyticdb プラグインは、データをバッチで書き込むことでこの問題に対処し、より低い CPU オーバーヘッドで 5 倍高いスループットを実現します。

前提条件

開始する前に、以下のものが用意されていることを確認してください。

Logstash がインストールされていること。詳細については、「Installing Logstash」をご参照ください。
AnalyticDB for MySQL の接続情報：ホスト、ポート、データベース名、ユーザー名、パスワード。

プラグインのインストール

データを書き込む前に、プラグインを最新バージョンにアップグレードして、最新のバグ修正とパフォーマンス改善を適用することを推奨します。

bin/logstash-plugin update logstash-output-analyticdb

Logstash のルートディレクトリに移動します。
```
cd logstash
```
logstash-output-analyticdb プラグインをインストールします。
```
bin/logstash-plugin install logstash-output-analyticdb
```

JDBC ドライバーディレクトリを作成し、コネクターの JAR をダウンロードします。

mkdir -p vendor/jar/jdbc
cd vendor/jar/jdbc
wget http://central.maven.org/maven2/mysql/mysql-connector-java/5.1.36/mysql-connector-java-5.1.36.jar

プラグインの設定

config ディレクトリに、設定ファイル「logstash-analyticdb.conf」を作成します (別の名前も使用可能です)。

次の例では、stdin から読み取り、AnalyticDB for MySQL に書き込みます。

input {
    stdin { }
}
output {
    analyticdb {
        driver_class => "com.mysql.jdbc.Driver"
        connection_string => "jdbc:mysql://<hostname>:<port>/<database>?user=<username>&password=<password>"
        statement => [ "INSERT INTO log (host, timestamp, message) VALUES(?, ?, ?)", "host", "@timestamp", "message" ]
        commit_size => 4194304
    }
}

プレースホルダーを実際の値に置き換えてください。

プレースホルダー	説明
`<hostname>`	AnalyticDB for MySQL のエンドポイント
`<port>`	ポート番号
`<database>`	対象データベース名
`<username>`	データベースのユーザー名
`<password>`	データベースのパスワード

プラグインパラメーター

パラメーター	タイプ	必須	デフォルト	説明
`driver_class`	文字列	はい	—	JDBC ドライバークラス。 MySQL 互換のエンドポイントには `com.mysql.jdbc.Driver` を使用します。
`connection_string`	文字列	はい	—	AnalyticDB for MySQL の JDBC 接続 URL。
`statement`	配列	はい	—	配列形式の INSERT ステートメントです。最初の要素は SQL テンプレートで、後続の要素は `?` プレースホルダーに対応するフィールド名です。
`commit_size`	整数	いいえ	なし	バッチあたりの最大データ量 (バイト単位)。この制限または `flush_size` のいずれかに先に達した時点で書き込みがトリガーされます。
`flush_size`	整数	いいえ	なし	バッチあたりの最大レコード数。この制限または `commit_size` のいずれかに先に達した時点で書き込みがトリガーされます。
`max_flush_exceptions`	整数	いいえ	100	書き込み例外が発生した場合の最大再試行回数。
`skip_exception`	ブール値	いいえ	`false`	すべての再試行に失敗した後の動作を制御します。`false` の場合、プログラムは例外をスローして終了します。`true` の場合、例外はスキップされ、ログに書き込まれます。

利用可能なすべてのパラメーターについては、プラグインの README をご参照ください。 Logstash の一般的な設定オプションについては、Logstash のドキュメントをご参照ください。

タスクの開始

Logstash のインストールディレクトリから、次のコマンドを実行します。

bin/logstash -f config/logstash-analyticdb.conf

Logstash は設定された入力からデータの読み取りを開始し、バッチで AnalyticDB for MySQL にデータを書き込みます。