Elasticsearch:Gunakan Canal untuk menyinkronkan data MySQL ke Alibaba Cloud Elasticsearch

Cara kerja

Canal bertindak sebagai node sekunder dari instance ApsaraDB RDS for MySQL Anda dan berlangganan log binernya. Saat baris data dimasukkan, diperbarui, atau dihapus di MySQL, Canal menangkap setiap event perubahan tersebut dan menuliskannya ke Elasticsearch melalui RESTful API.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

• Instance ApsaraDB RDS for MySQL. Lihat Buat instance ApsaraDB RDS for MySQL. Panduan ini menggunakan instance MySQL 5.7.

• Kluster Alibaba Cloud Elasticsearch. Lihat Buat kluster Alibaba Cloud Elasticsearch. Panduan ini menggunakan kluster Elasticsearch V6.7 (Edisi Kernel-enhanced).

• Instance ECS yang menjalankan CentOS 7.6 (64-bit), digunakan untuk meng-host Canal server dan Canal adapter. Lihat Buat instance menggunakan wizard.

• Ketiga sumber daya tersebut berada dalam virtual private cloud (VPC) yang sama.

• Alamat IP instance ECS ditambahkan ke daftar putih alamat IP kluster Elasticsearch. Lihat Konfigurasi daftar putih alamat IP publik atau privat untuk kluster Elasticsearch.

Batasan

• Canal hanya menyinkronkan data inkremental — ekspor tabel penuh tidak didukung.

• Canal adapter tidak dapat terhubung ke kluster Elasticsearch melalui HTTPS.

• Nama bidang dan tipe data dalam pemetaan indeks Elasticsearch harus persis sesuai dengan kolom tabel MySQL yang bersangkutan.

• Versi Java Development Kit (JDK) harus 1.8.0 atau lebih baru.

• Kompatibilitas versi Canal dan Elasticsearch:

  Catatan Canal 1.1.4 tidak dapat digunakan untuk menyinkronkan data ke kluster Elasticsearch V7.X. Gunakan Canal 1.1.5 untuk kluster Elasticsearch V7.x, dan Canal 1.1.7 untuk kluster Elasticsearch V8.x. Panduan ini menggunakan Canal 1.1.4 dengan kluster Elasticsearch V6.7.

  Versi Canal	Versi Elasticsearch yang didukung
  1.1.4	6.x
  1.1.5	7.x
  1.1.7	8.x

Langkah 1: Siapkan sumber data MySQL

Masuk ke Konsol ApsaraDB RDS dan buat database serta tabel. Panduan ini menggunakan skema tabel berikut:

CREATE TABLE `es_test` (
    `id`    bigint(32) NOT NULL,
    `name`  text NOT NULL,
    `count` text NOT NULL,
    `color` text NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

Untuk instruksi umum, lihat Alur kerja umum penggunaan ApsaraDB RDS for MySQL.

Langkah 2: Buat indeks Elasticsearch dengan pemetaan

Nama bidang dan tipe data dalam pemetaan indeks harus persis sesuai dengan tabel MySQL.

1. Masuk ke Konsol Kibana kluster Elasticsearch Anda. Lihat Masuk ke Konsol Kibana.

   Catatan Panduan ini menggunakan Elasticsearch V6.7.0. Operasi mungkin berbeda pada versi lain.

2. Di panel navigasi kiri, klik Dev Tools.

3. Pada tab Console, jalankan perintah berikut untuk membuat indeks es_test:

   PUT es_test?include_type_name=true
   {
     "settings": {
       "index": {
         "number_of_shards": "5",
         "number_of_replicas": "1"
       }
     },
     "mappings": {
       "_doc": {
         "properties": {
           "count": {
             "type": "text"
           },
           "id": {
             "type": "integer"
           },
           "name": {
             "type": "text",
             "analyzer": "ik_smart"
           },
           "color": {
             "type": "text"
           }
         }
       }
     }
   }

   Tanggapan sukses akan tampak seperti:

   {
     "acknowledged": true,
     "shards_acknowledged": true,
     "index": "es_test"
   }

Langkah 3: Instal JDK

1. Hubungkan ke instance ECS. Lihat Hubungkan ke instance Linux menggunakan password atau kunci.

   Catatan Dalam contoh ini, digunakan pengguna biasa.

2. Cari paket JDK yang tersedia:

   sudo yum search java | grep -i --color JDK

3. Instal JDK. Panduan ini menggunakan java-1.8.0-openjdk-devel.x86_64:

   sudo yum install java-1.8.0-openjdk-devel.x86_64

4. Konfigurasi variabel lingkungan:

   1. Buka ~/.bash_profile:

      vim ~/.bash_profile

   2. Tambahkan baris berikut. Ganti JAVA_HOME dengan jalur instalasi JDK aktual Anda. Jalankan find / -name 'java' untuk menemukannya.

      export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.362.b08-1.el7_9.x86_64
      export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
      export PATH=$PATH:$JAVA_HOME/bin

   3. Tekan Esc, ketik :wq untuk menyimpan, lalu terapkan konfigurasi:

      source ~/.bash_profile

5. Verifikasi instalasi:

   java -version

   Output yang diharapkan:

   openjdk version "1.8.0_362"
   OpenJDK Runtime Environment (build 1.8.0_362-b08)
   OpenJDK 64-Bit Server VM (build 25.362-b08, mixed mode)

Langkah 4: Instal dan mulai Canal server

1. Unduh Canal server 1.1.4:

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz

   Untuk versi lain, lihat Rilis Canal.

2. Ekstrak paket:

   tar -zxvf canal.deployer-1.1.4.tar.gz

3. Edit conf/example/instance.properties:

   Parameter	Deskripsi	Contoh
   canal.instance.master.address	Titik akhir internal dan port instance ApsaraDB RDS for MySQL. Dapatkan informasi ini dari halaman Basic Information instance RDS.	rm-bp1u1xxxxxxxxx6ph.mysql.rds.aliyuncs.com:3306
   canal.instance.dbUsername	Username database. Dapatkan dari halaman Accounts instance RDS.	canal_user
   canal.instance.dbPassword	Password database.	<your-password>

   vi conf/example/instance.properties

   Atur parameter berikut:

   

4. Tekan Esc, ketik :wq untuk menyimpan.

5. Jalankan Canal server dan verifikasi log:

   ./bin/startup.sh
   cat logs/canal/canal.log

   

Langkah 5: Instal dan mulai Canal adapter

1. Unduh Canal adapter 1.1.4:

   wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.adapter-1.1.4.tar.gz

2. Ekstrak paket:

   tar -zxvf canal.adapter-1.1.4.tar.gz

3. Edit conf/application.yml:

   Parameter	Deskripsi	Contoh
   canal.conf.canalServerHost	Alamat Canal server. Gunakan nilai default.	127.0.0.1:11111
   canal.conf.srcDataSources.defaultDS.url	URL koneksi JDBC instance ApsaraDB RDS for MySQL. Gunakan format jdbc:mysql://<endpoint>:<port>/<database>?useUnicode=true. Dapatkan endpoint dan port dari halaman Basic Information.	jdbc:mysql://rm-bp1xxxxxxxxxnd6ph.mysql.rds.aliyuncs.com:3306/elasticsearch?useUnicode=true
   canal.conf.srcDataSources.defaultDS.username	Username database. Dapatkan dari halaman Accounts.	canal_user
   canal.conf.srcDataSources.defaultDS.password	Password database.	<your-password>
   canal.conf.canalAdapters.groups.outerAdapters.hosts (di bawah name: es)	Titik akhir internal dan port kluster Elasticsearch. Dapatkan dari halaman Basic Information kluster.	es-cn-v64xxxxxxxxx3medp.elasticsearch.aliyuncs.com:9200
   canal.conf.canalAdapters.groups.outerAdapters.mode	Mode transport. Atur ke rest.	rest
   canal.conf.canalAdapters.groups.outerAdapters.properties.security.auth	Kredensial Elasticsearch dalam format <username>:<password>.	elastic:es_password
   canal.conf.canalAdapters.groups.outerAdapters.properties.cluster.name	ID kluster Elasticsearch. Dapatkan dari halaman Basic Information.	es-cn-v64xxxxxxxxx3medp

   vi conf/application.yml

   Atur parameter berikut:

   

4. Tekan Esc, ketik :wq untuk menyimpan.

5. Edit file pemetaan bidang conf/es/*.yml untuk menentukan cara kolom MySQL dipetakan ke bidang Elasticsearch:

   Parameter	Deskripsi	Nilai yang digunakan dalam panduan ini
   esMapping._index	Nama indeks Elasticsearch yang dibuat di Langkah 2.	es_test
   esMapping._type	Tipe indeks Elasticsearch.	_doc
   esMapping._id	Bidang ID dokumen.	_id
   esMapping.sql	Kueri SQL yang memilih kolom untuk disinkronkan.	select t.id as _id, t.id, t.count, t.name, t.color from es_test t

   

6. Jalankan Canal adapter dan verifikasi log:

   Catatan Canal adapter 1.1.4 menyertakan driver MySQL versi 5.1.40. Jika instance ApsaraDB RDS Anda menjalankan MySQL 8.x, adapter gagal dijalankan. Lihat Ganti driver MySQL untuk MySQL 8.x di bagian Pemecahan Masalah.

   ./bin/startup.sh
   cat logs/adapter/adapter.log

   

Langkah 6: Verifikasi sinkronisasi

1. Di database ApsaraDB RDS for MySQL, masukkan satu baris ke tabel es_test:

   INSERT INTO `ES`.`es_test` (`count`, `id`, `name`, `color`)
   VALUES ('11', 2, 'canal_test2', 'red');

   Anda juga dapat menguji dengan pernyataan UPDATE atau DELETE.

2. Masuk ke Konsol Kibana kluster Elasticsearch Anda. Lihat Masuk ke Konsol Kibana.

3. Di panel navigasi kiri, klik Dev Tools.

4. Pada tab Console, jalankan kueri berikut:

   Penting

   Canal hanya menyinkronkan perubahan inkremental.

   GET /es_test/_search

   Jika sinkronisasi berfungsi, tanggapan mencakup baris yang Anda masukkan:

   

Pemecahan Masalah

Pada contoh berikut, digunakan pengguna root.

java.lang.RuntimeException: java.lang.RuntimeException: java.lang.ClassCastException:
com.alibaba.druid.pool.DruidDataSource cannot be cast to
com.alibaba.druid.pool.DruidDataSource
at com.alibaba.otter.canal.client.adapter.es7x.ES7xAdapter.init(ES7xAdapter.java:54)
~[client-adapter.es7x-1.1.5-jar-with-dependencies.jar:na]

Langkah selanjutnya

• Konfigurasi daftar putih alamat IP publik atau privat untuk kluster Elasticsearch

• Masuk ke Konsol Kibana

• Lihat status kluster dan informasi node

• Konfigurasi sinonim

• Plug-in