Elasticsearch:Gunakan Logstash untuk menyinkronkan data dari PolarDB-X (DRDS) ke Elasticsearch

Buat tabel di PolarDB-X 1.0 dan masukkan data uji. Contoh berikut membuat tabel food:

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

Kolom utama:

• id: Primary key. Nilainya dipetakan ke bidang _id di Elasticsearch. Diperlukan agar pipeline sinkronisasi dapat mengidentifikasi dan menimpa dokumen dengan benar.

• update_time: Timestamp pelacakan. Plug-in logstash-input-jdbc menggunakan bidang ini untuk mendeteksi perubahan sejak pengambilan terakhir, sehingga sinkronisasi inkremental dapat berfungsi.

• insert_time: Mencatat kapan baris pertama kali dibuat. Tidak wajib untuk sinkronisasi, tetapi berguna untuk audit.

Masukkan data sampel:

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());

Aktifkan fitur Auto Indexing untuk kluster Elasticsearch Anda. Untuk detailnya, lihat Access and configure an Elasticsearch cluster.

Unggah driver JDBC MySQL ke kluster Logstash. Versi driver harus kompatibel dengan instans PolarDB-X 1.0 Anda. Contoh ini menggunakan mysql-connector-java-5.1.35. Untuk instruksi pengunggahan, lihat Configure third-party libraries.

Tambahkan alamat IP node kluster Logstash ke daftar putih alamat IP instans PolarDB-X 1.0 Anda. Temukan alamat IP node di halaman Basic Information kluster Logstash di Konsol Elasticsearch. Untuk instruksi pendaftaran daftar putih, lihat Set an IP address whitelist.

Buka halaman Logstash Clusters di Konsol Alibaba Cloud Elasticsearch.

Di bilah navigasi atas, pilih wilayah tempat kluster Anda berada.

Di halaman Logstash Clusters, temukan kluster Anda dan klik ID-nya.

Di panel navigasi kiri, klik Pipelines.

Di halaman Pipelines, klik Create Pipeline.

Di halaman Create Task, masukkan ID pipeline di bidang Pipeline ID, lalu masukkan konfigurasi berikut di bidang Config Settings. Ganti <Logstash cluster ID>, <Database name>, dan kredensial dengan nilai aktual Anda. Untuk menemukan ID kluster Logstash Anda, lihat Overview of the Logstash Clusters page.

Parameter Masukan Plugin

Output Plugin Parameters

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}"
  }
}

Untuk daftar lengkap parameter plug-in input, lihat Logstash JDBC input plugin. Untuk sintaks konfigurasi pipeline umum, lihat Logstash configuration files.

Klik Next untuk mengonfigurasi parameter pipeline.

Klik Save atau Save and Deploy.

• Save: Menyimpan konfigurasi pipeline dan memicu perubahan kluster, tetapi tidak mengaktifkan pipeline. Untuk mengaktifkannya, buka halaman Pipelines, temukan pipeline tersebut, lalu klik Deploy di kolom Actions.

• Save and Deploy: Menyimpan dan segera merestart kluster Logstash untuk mengaktifkan pipeline.

Login ke Konsol Kibana kluster Elasticsearch. Untuk detailnya, lihat Log on to the Kibana console.

Di panel navigasi kiri, klik Dev Tools.

Di tab Console, jalankan kueri berikut untuk memastikan tiga catatan sampel telah disinkronkan:

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

Tanggapan yang diharapkan:

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

Perbarui catatan yang ada dan masukkan catatan baru di PolarDB-X 1.0:

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

Setelah siklus polling berikutnya selesai (dalam satu menit), verifikasi pembaruan di Kibana. Kueri catatan yang diperbarui:

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

Jika perintah berhasil dijalankan, hasil seperti pada gambar berikut akan ditampilkan. Kueri semua catatan untuk memastikan baris baru telah ditambahkan:

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

Jika perintah berhasil dijalankan, hasil seperti pada gambar berikut akan ditampilkan.