Elasticsearch:Logstash を使用した PolarDB-X (DRDS) から Elasticsearch へのデータ同期

PolarDB-X 1.0 でテーブルを作成し、テストデータを挿入します。次の例では、food テーブルを作成します。

CREATE TABLE food(
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(32),
  insert_time DATETIME,
  update_time DATETIME
);

主要な列：

• id：プライマリキー。その値は Elasticsearch の _id フィールドにマッピングされます。同期パイプラインがドキュメントを正しく識別して上書きするために必要です。

• update_time：追跡タイムスタンプ。logstash-input-jdbc プラグインは、このフィールドを使用して前回のポーリング以降の変更を検出します。増分同期が機能するために必要です。

• insert_time：行が最初に作成された日時を記録します。同期には必要ありませんが、監査に役立ちます。

サンプルデータを挿入します。

INSERT INTO food VALUES(null, 'Chocolates', NOW(), NOW());
INSERT INTO food VALUES(null, 'Yogurt', NOW(), NOW());
INSERT INTO food VALUES(null, 'Ham sausage', NOW(), NOW());

ご利用の Elasticsearch クラスターの自動インデックス作成機能を有効にします。詳細については、「Elasticsearch クラスターへのアクセスと設定」をご参照ください。

MySQL JDBC ドライバーを Logstash クラスターにアップロードします。ドライバーのバージョンは、ご利用の PolarDB-X 1.0 インスタンスと互換性がある必要があります。この例では、mysql-connector-java-5.1.35 を使用します。アップロード手順については、「サードパーティライブラリの設定」をご参照ください。

Logstash クラスターノードの IP アドレスを、ご利用の PolarDB-X 1.0 インスタンスの IP アドレスホワイトリストに追加します。ノードの IP アドレスは、Elasticsearch コンソールの Logstash クラスターの [基本情報] ページで確認できます。ホワイトリストへの追加手順については、「IP アドレスホワイトリストの設定」をご参照ください。

Alibaba Cloud Elasticsearch コンソールの Logstash クラスターページに移動します。

上部のナビゲーションバーで、クラスターが存在するリージョンを選択します。

[Logstash クラスター] ページで、対象のクラスターを見つけてその ID をクリックします。

左側のナビゲーションウィンドウで、[パイプライン] をクリックします。

[パイプライン] ページで、[パイプラインの作成] をクリックします。

[タスクの作成] ページで、[パイプライン ID] フィールドにパイプライン ID を入力し、[設定] フィールドに次の構成を入力します。<Logstash cluster ID>、<Database name>、および認証情報を実際の値に置き換えてください。Logstash クラスター ID を確認するには、「Logstash クラスターページの概要」をご参照ください。

Input プラグインパラメーター

Output プラグインパラメーター

input {
  jdbc {
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_driver_library => "/ssd/1/share/<Logstash cluster ID>/logstash/current/config/custom/mysql-connector-java-5.1.35.jar"
    jdbc_connection_string => "jdbc:mysql://drdshbga51x6****.drds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    statement => "select * from food where update_time >= :sql_last_value"
    schedule => "* * * * *"
    record_last_run => true
    last_run_metadata_path => "/ssd/1/<Logstash cluster ID>/logstash/data/last_run_metadata_update_time.txt"
    clean_run => false
    tracking_column_type => "timestamp"
    use_column_value => true
    tracking_column => "update_time"
  }
}
filter {

}
output {
  elasticsearch {
    hosts => "http://es-cn-n6w1o1x0w001c****.elasticsearch.aliyuncs.com:9200"
    user => "elastic"
    password => "es_password"
    index => "drds_test"
    document_id => "%{id}"
  }
}

Input プラグインパラメーターの完全なリストについては、「Logstash JDBC input plugin」をご参照ください。一般的なパイプライン構成の構文については、「Logstash 構成ファイル」をご参照ください。

[次へ] をクリックして、パイプラインパラメーターを設定します。

[保存] または [保存してデプロイ] をクリックします。

• 保存：パイプライン構成を保存し、クラスターの変更をトリガーしますが、パイプラインはアクティブ化されません。アクティブ化するには、[パイプライン] ページに移動し、パイプラインを見つけて、[操作] 列の [デプロイ] をクリックします。

• 保存してデプロイ：Logstash クラスターを保存し、すぐに再起動してパイプラインをアクティブ化します。

Elasticsearch クラスターの Kibana コンソールにログインします。詳細については、「Kibana コンソールへのログイン」をご参照ください。

左側のナビゲーションウィンドウで、[開発ツール] をクリックします。

[コンソール] タブで、次のクエリを実行して、3 つのサンプルレコードが同期されたことを確認します。

GET drds_test/_count
{
  "query": {"match_all": {}}
}

期待される応答：

{
  "count": 3,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  }
}

PolarDB-X 1.0 で既存のレコードを更新し、新しいレコードを挿入します。

UPDATE food SET name='Chocolates', update_time=NOW() WHERE id = 1;
INSERT INTO food VALUES(null, 'Egg', NOW(), NOW());

次のポーリングサイクルが完了した後 (1 分以内)、Kibana で更新を確認します。更新されたレコードをクエリします。

GET drds_test/_search
{
  "query": {
    "match": {
      "name": "Chocolates"
    }
  }
}

コマンドが正常に実行されると、次の図に示す結果が返されます。 すべてのレコードをクエリして、新しい行が追加されたことを確認します。

GET drds_test/_search
{
  "query": {
    "match_all": {}
  }
}

コマンドが正常に実行されると、次の図に示す結果が返されます。