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

Informasi latar belakang

Alibaba Cloud Logstash adalah alat pemrosesan data yang kuat yang dapat digunakan untuk mengumpulkan, mentransformasi, mengoptimalkan, dan menghasilkan data. Alibaba Cloud Logstash menyediakan plug-in logstash-input-jdbc, yang diinstal secara default pada kluster Logstash dan tidak dapat dihapus. Plug-in ini dapat meminta beberapa catatan data dalam ApsaraDB RDS for MySQL sekaligus dan menyinkronkannya ke Alibaba Cloud Elasticsearch. Plug-in ini menggunakan metode round-robin untuk mengidentifikasi catatan data yang baru saja dimasukkan atau diperbarui dalam ApsaraDB RDS for MySQL secara berkala dan menyinkronkannya ke Alibaba Cloud Elasticsearch. Untuk informasi lebih lanjut, lihat Gunakan Logstash dan JDBC untuk menyinkronkan data dari database relasional ke kluster Elasticsearch. Jika Anda ingin menyinkronkan data penuh dengan latensi beberapa detik atau meminta data tertentu pada satu waktu, Anda dapat menggunakan Alibaba Cloud Logstash.

Prasyarat

Disarankan untuk membuat instance dan kluster berikut dalam virtual private cloud (VPC) yang sama.

• Buat instance ApsaraDB RDS for MySQL. Untuk informasi lebih lanjut, lihat Buat instance ApsaraDB RDS for MySQL. Dalam contoh ini, instance ApsaraDB RDS yang menjalankan MySQL 5.7 dibuat.

• Buat kluster Alibaba Cloud Elasticsearch. Untuk informasi lebih lanjut, lihat Buat kluster Alibaba Cloud Elasticsearch. Dalam contoh ini, kluster Elasticsearch V7.10 dibuat.

• Buat kluster Alibaba Cloud Logstash. Untuk informasi lebih lanjut, lihat Buat kluster Alibaba Cloud Logstash.

Batasan

Prosedur

Tanya Jawab Umum

• P: Apa yang harus saya lakukan jika tugas sinkronisasi data saya gagal karena pipeline macet dalam status menginisialisasi, data sebelum dan sesudah sinkronisasi tidak konsisten, atau koneksi ke database gagal?

  A: Periksa apakah log kluster Logstash Anda berisi informasi kesalahan dan identifikasi penyebabnya berdasarkan informasi kesalahan tersebut. Untuk informasi lebih lanjut, lihat Kueri log. Tabel berikut menjelaskan penyebab umum kesalahan dan solusi untuk kesalahan tersebut.

  Catatan

  Jika operasi pembaruan sedang dilakukan pada kluster Logstash Anda saat Anda melakukan operasi yang dijelaskan dalam solusi berikut, jeda operasi pembaruan dengan merujuk ke Lihat kemajuan tugas kluster. Setelah operasi yang dijelaskan dalam solusi selesai, sistem me-restart kluster Logstash dan melanjutkan operasi pembaruan.

  Penyebab	Solusi
  Alamat IP node dalam kluster Logstash belum ditambahkan ke daftar putih instance ApsaraDB RDS for MySQL.	Tambahkan alamat IP node dalam kluster Logstash ke daftar putih instance ApsaraDB RDS for MySQL. Untuk informasi lebih lanjut, lihat Gunakan klien database atau CLI untuk terhubung ke instance ApsaraDB RDS for MySQL. Catatan Untuk informasi lebih lanjut tentang cara mendapatkan alamat IP node dalam kluster Logstash, lihat Lihat informasi dasar kluster.
  Anda menggunakan kluster Logstash untuk menyinkronkan data dari database MySQL yang dikelola sendiri yang di-hosting pada instance ECS, tetapi alamat IP privat dan port internal node dalam kluster belum ditambahkan ke grup keamanan instance ECS.	Tambahkan alamat IP privat dan port internal node dalam kluster ke grup keamanan instance ECS. Untuk informasi lebih lanjut, lihat Tambahkan aturan grup keamanan. Catatan Untuk informasi lebih lanjut tentang cara mendapatkan alamat IP dan port node dalam kluster Logstash, lihat Lihat informasi dasar kluster.
  Kluster Elasticsearch tidak berada dalam VPC yang sama dengan kluster Logstash.	Gunakan salah satu solusi berikut: Beli kluster Elasticsearch yang berada dalam VPC yang sama dengan kluster Logstash. Untuk informasi lebih lanjut, lihat Buat kluster Alibaba Cloud Elasticsearch. Setelah kluster Elasticsearch dibeli, modifikasi konfigurasi pipeline kluster Logstash. Konfigurasikan gateway NAT untuk transmisi data melalui Internet. Untuk informasi lebih lanjut, lihat Konfigurasikan gateway NAT untuk transmisi data melalui Internet.
  Endpoint instance ApsaraDB RDS for MySQL salah, dan nomor port instance bukan 3306.	Dapatkan endpoint dan nomor port yang benar. Untuk informasi lebih lanjut, lihat Lihat dan kelola endpoint dan port instance. Kemudian, ganti endpoint dan nomor port dalam nilai parameter jdbc_connection_string dengan endpoint dan nomor port yang Anda dapatkan. Penting <Endpoint instance ApsaraDB RDS for MySQL>: Anda harus menentukan endpoint internal instance ApsaraDB RDS for MySQL. Jika Anda ingin menggunakan endpoint publik instance ApsaraDB RDS for MySQL, Anda harus mengonfigurasi gateway NAT untuk kluster Logstash untuk mengaktifkan kluster Logstash terhubung ke Internet. Untuk informasi lebih lanjut, lihat Konfigurasikan gateway NAT untuk transmisi data melalui Internet.
  Fitur Auto Indexing dinonaktifkan untuk kluster Elasticsearch.	Aktifkan fitur Auto Indexing untuk kluster Elasticsearch. Untuk informasi lebih lanjut, lihat Konfigurasikan file YML.
  Beban kluster Elasticsearch atau Logstash terlalu tinggi.	Tingkatkan konfigurasi kluster Elasticsearch atau Logstash. Untuk informasi lebih lanjut, lihat Tingkatkan konfigurasi kluster. Catatan Jika beban kluster Elasticsearch terlalu tinggi, Anda dapat melihat data pemantauan yang dikumpulkan berdasarkan metrik di konsol Elasticsearch untuk mendapatkan informasi beban kluster Elasticsearch. Untuk informasi lebih lanjut, lihat Metrik dan saran penanganan anomali. Jika beban kluster Logstash terlalu tinggi, Anda dapat mengaktifkan fitur X-Pack Monitoring untuk kluster Logstash, gunakan fitur tersebut untuk memantau kluster Logstash, dan kemudian lihat data pemantauan. Untuk informasi lebih lanjut, lihat Aktifkan fitur X-Pack Monitoring.
  Tidak ada file driver yang digunakan untuk koneksi berbasis JDBC ke database ApsaraDB RDS for MySQL yang diunggah.	Unggah file driver. Untuk informasi lebih lanjut, lihat Konfigurasikan pustaka pihak ketiga.
  file_extend ditentukan dalam konfigurasi pipeline. Namun, plug-in logstash-output-file_extend belum diinstal.	Gunakan salah satu solusi berikut: Instal plug-in logstash-output-file_extend. Untuk informasi lebih lanjut, lihat Instal dan hapus plug-in. Hapus parameter file_extend dari konfigurasi pipeline.

  Untuk informasi lebih lanjut tentang penyebab dan solusi untuk masalah ini, lihat FAQ tentang transfer data menggunakan Logstash.

• P: Bagaimana cara mengonfigurasi beberapa koneksi JDBC ke database sumber dalam konfigurasi input pipeline?

  A: Anda dapat mendefinisikan beberapa sumber data JDBC dalam konfigurasi input pipeline dan menentukan pernyataan query untuk setiap tabel dalam parameter statement. Kode berikut memberikan contoh:

  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/<Nama Database>?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/<Nama Database>?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}"
          }
      }
  }

  Dalam contoh sebelumnya, parameter type ditambahkan ke konfigurasi jdbc. Parameter ini membantu Anda mendefinisikan kondisi dalam konfigurasi output dan menyinkronkan data dari tabel yang berbeda ke indeks yang berbeda.