Elasticsearch:Gunakan Logstash untuk menyinkronkan data dari ApsaraDB RDS for MySQL ke Elasticsearch

Batasan

• Pastikan instans ApsaraDB RDS for MySQL, kluster Alibaba Cloud Logstash, dan kluster Alibaba Cloud Elasticsearch berada dalam zona waktu yang sama. Jika tidak, saat data terkait waktu disinkronkan, akan terjadi perbedaan zona waktu antara data sebelum dan sesudah sinkronisasi.

• Field _id di Elasticsearch harus sama dengan field id di MySQL.

  Kondisi ini memastikan bahwa ketika sebuah record di MySQL ditulis ke Elasticsearch, tugas sinkronisasi dapat membuat pemetaan langsung antara record MySQL dan dokumen Elasticsearch. Misalnya, ketika sebuah record diperbarui di MySQL, tugas sinkronisasi akan menimpa dokumen Elasticsearch yang memiliki ID yang sama dengan record yang diperbarui tersebut.

  Catatan

  Berdasarkan mekanisme internal Elasticsearch, pembaruan pada dasarnya merupakan penghapusan dokumen lama diikuti dengan pengindeksan dokumen baru. Oleh karena itu, menimpa dokumen di Elasticsearch sama efisiennya dengan operasi update.

• Saat data dimasukkan atau diperbarui di MySQL, record yang bersangkutan harus berisi field yang menyimpan waktu pembaruan atau waktu penyisipan.

  Setiap kali Logstash melakukan polling ke MySQL, Logstash menyimpan waktu pembaruan atau waktu penyisipan dari record terakhir yang dibacanya dari MySQL. Saat membaca data, Logstash hanya membaca record yang memenuhi kondisi tersebut, yaitu record yang waktu pembaruan atau waktu penyisipannya lebih baru daripada record terakhir pada polling sebelumnya.

  Penting

  Plugin logstash-input-jdbc tidak dapat menyinkronkan penghapusan. Anda harus menjalankan perintah terkait di Elasticsearch untuk menghapus dokumen secara manual.

Prasyarat

Kami menyarankan Anda membuat instans berikut dalam virtual private cloud (VPC) yang sama:

Anda juga dapat menggunakan layanan yang dideploy di Internet. Dalam kasus ini, Anda harus mengonfigurasi SNAT, mengaktifkan titik akhir publik instans ApsaraDB RDS for MySQL, dan menghapus batasan daftar putih. Untuk informasi lebih lanjut tentang cara mengonfigurasi SNAT, lihat Konfigurasi gerbang NAT untuk transfer data publik. Untuk informasi lebih lanjut tentang cara mengonfigurasi daftar putih, lihat Konfigurasi daftar putih alamat IP.

• Buat instans ApsaraDB RDS for MySQL. Untuk informasi lebih lanjut, lihat Buat instans ApsaraDB RDS for MySQL. Pada topik ini, digunakan MySQL 5.7.

• Buat kluster Alibaba Cloud Elasticsearch. Untuk informasi lebih lanjut, lihat Buat kluster Alibaba Cloud Elasticsearch. Pada topik ini, digunakan kluster Elasticsearch V8.17.

• Buat kluster Alibaba Cloud Logstash. Untuk informasi lebih lanjut, lihat Buat instans Alibaba Cloud Logstash. Pada topik ini, digunakan kluster Logstash V8.11.4.

Informasi latar belakang

Alibaba Cloud Logstash adalah tool pengumpulan dan pemrosesan data yang menyediakan kemampuan pengumpulan data, transformasi, optimasi, dan output. Plugin logstash-input-jdbc Logstash telah diinstal secara default dan tidak dapat diuninstall. Anda dapat menggunakan plugin ini untuk mengkueri data di instans ApsaraDB RDS for MySQL secara batch dan menyinkronkan data tersebut ke Elasticsearch. Plugin ini juga secara berkala melakukan polling terhadap data di instans ApsaraDB RDS for MySQL dan menyinkronkan record yang disisipkan atau diubah sejak polling sebelumnya ke Elasticsearch. Untuk informasi lebih lanjut, lihat How to keep Elasticsearch synchronized with a relational database using Logstash dalam dokumentasi resmi. Solusi ini cocok untuk skenario di mana data penuh disinkronkan dan latensi beberapa detik dapat diterima, atau skenario di mana data yang memenuhi kondisi tertentu dikueri secara batch lalu disinkronkan.

Sinkronisasi data

FAQ

Pipeline saya macet di status initializing, data tidak konsisten setelah sinkronisasi, atau koneksi database gagal. Apa yang harus saya lakukan?

Periksa log kluster Logstash terlebih dahulu. Buka konsol Logstash dan gunakan fitur Query logs untuk melihat detail error. Untuk detailnya, lihat Kueri log.

Jika pembaruan kluster sedang berlangsung saat Anda menerapkan perbaikan, jeda pembaruan tersebut terlebih dahulu. Lihat Lihat progres tugas kluster. Setelah perbaikan, sistem akan merestart kluster dan melanjutkan pembaruan secara otomatis.

Tabel berikut mencantumkan penyebab umum dan solusinya:

Untuk panduan troubleshooting lebih lanjut, lihat FAQ tentang transfer data menggunakan Logstash.

Bagaimana cara menyinkronkan data dari beberapa tabel MySQL ke indeks Elasticsearch yang terpisah?

Definisikan beberapa blok jdbc dalam bagian input, tetapkan nilai type untuk masing-masing, lalu gunakan kondisi if[type] dalam bagian output untuk mengarahkan data setiap tabel ke indeks yang berbeda:

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.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableA 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"
      type => "A"
    }
    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.48.jar"
      jdbc_connection_string => "jdbc:mysql://rm-bp1xxxxx.mysql.rds.aliyuncs.com:3306/<Database name>?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowLoadLocalInfile=false&autoDeserialize=false"
      jdbc_user => "xxxxx"
      jdbc_password => "xxxx"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement => "select * from tableB 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"
      type => "B"
    }
}
output {
    if[type] == "A" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_A"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
    if[type] == "B" {
        elasticsearch {
            hosts => "http://es-cn-0h****dd0hcbnl.elasticsearch.aliyuncs.com:9200"
            index => "rds_es_dxhtest_datetime_B"
            user => "elastic"
            password => "xxxxxxx"
            document_id => "%{id}"
        }
    }
}