Sinkronkan data real-time ke Elasticsearch menggunakan DTS dan Alibaba Cloud Elasticsearch - Elasticsearch

Gunakan Data Transmission Service (DTS) untuk menyinkronkan data dari database ApsaraDB RDS for MySQL ke kluster Alibaba Cloud Elasticsearch secara real time. Untuk informasi selengkapnya tentang DTS, lihat DTS. DTS mendukung sinkronisasi perubahan data yang dihasilkan oleh operasi INSERT, DELETE, dan UPDATE, sehingga pendekatan ini cocok untuk kasus penggunaan yang memerlukan pencarian atau analitik berlatensi rendah pada data MySQL produksi.

Tutorial ini memandu Anda melalui seluruh proses penyiapan: menyiapkan database sumber dan kluster tujuan, membuat tugas sinkronisasi, serta memverifikasi hasil sinkronisasi penuh maupun inkremental.

Cara kerja

DTS berjalan dalam dua fase:

Full data sync — DTS membaca baris yang sudah ada dari tabel MySQL sumber dan menuliskannya ke Elasticsearch sebagai dokumen.
Incremental sync — Setelah sinkronisasi penuh selesai, DTS beralih ke change data capture dan terus-menerus menerapkan operasi INSERT, UPDATE, dan DELETE ke Elasticsearch.

Selama sinkronisasi penuh awal, DTS membaca dan menulis sumber daya di kedua sisi (sumber dan tujuan), yang dapat meningkatkan beban. Jalankan sinkronisasi selama jam sepi ketika utilisasi CPU di kedua sisi berada di bawah 30%. Jika Anda memulai sinkronisasi selama jam sibuk, sinkronisasi penuh mungkin gagal—restart tugas jika hal tersebut terjadi. Sinkronisasi inkremental selama jam sibuk dapat menyebabkan latensi.

DTS tidak menyinkronkan operasi DDL. Jika Anda mengubah skema tabel selama sinkronisasi, lihat Batasan untuk langkah pemulihan yang diperlukan.

Prasyarat

Sebelum memulai, pastikan Anda telah memiliki:

Instans ApsaraDB RDS for MySQL (MySQL 8.0 digunakan dalam tutorial ini)
Kluster Alibaba Cloud Elasticsearch (V7.10 digunakan dalam tutorial ini), dengan fitur Auto Indexing diaktifkan
Akses ke Konsol DTS

Batasan

Batasan	Detail
Operasi DDL tidak disinkronkan	Jika Anda menjalankan operasi DDL pada tabel sumber selama sinkronisasi, Anda harus: menghapus tabel tersebut dari tugas sinkronisasi, menghapus indeks yang sesuai dari kluster Elasticsearch, lalu menambahkan kembali tabel tersebut ke tugas. Lihat Menghapus objek dari tugas sinkronisasi data dan Menambahkan objek ke tugas sinkronisasi data.
Penambahan kolom memerlukan langkah manual	Untuk menambahkan kolom, pertama-tama perbarui mapping indeks di Elasticsearch agar mencakup bidang baru tersebut. Kemudian jalankan ALTER TABLE pada sumber, jeda tugas sinkronisasi, lalu restart tugas tersebut.
Perbedaan tipe data	ApsaraDB RDS for MySQL dan Elasticsearch menggunakan sistem tipe yang berbeda. Selama sinkronisasi skema awal, DTS memetakan bidang sumber ke bidang tujuan berdasarkan tipe Elasticsearch yang didukung. Lihat Pemetaan tipe data untuk sinkronisasi skema untuk tabel pemetaan lengkap.
Tabel tanpa primary key	Tabel tanpa primary key dapat menyebabkan DTS mengonsumsi lebih banyak sumber daya baca/tulis. Hal ini dapat meningkatkan beban pada sumber dan tujuan hingga titik di mana sumber atau tujuan menjadi tidak tersedia.

Langkah 1: Siapkan database sumber dan kluster tujuan

Siapkan database sumber

Buat instans ApsaraDB RDS yang menjalankan MySQL 8.0.
Buat akun dan database bernama test_mysql.

Di database test_mysql, buat tabel dan masukkan data sampel:

-- Buat tabel
CREATE TABLE `es_test` (
    `id` bigint(32) NOT NULL,
    `name` varchar(32) NULL,
    `age` bigint(32) NULL,
    `hobby` varchar(32) NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

-- Masukkan baris sampel
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (1,'user1',22,'music');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (2,'user2',23,'sport');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (3,'user3',43,'game');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (4,'user4',24,'run');
INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (5,'user5',42,'basketball');

Siapkan kluster Elasticsearch tujuan

Buat kluster Alibaba Cloud Elasticsearch yang menjalankan V7.10.
Aktifkan fitur Auto Indexing. Buka konfigurasi file YML kluster dan aktifkan Auto Indexing.

Langkah 2: Buat tugas sinkronisasi

Buka halaman Sinkronisasi Data di Konsol DTS.
Klik Create Task.
Konfigurasikan tugas sesuai petunjuk:

Untuk detail setiap parameter, lihat Sinkronkan data dari instans ApsaraDB RDS for MySQL ke kluster Elasticsearch.
1. Konfigurasikan sumber (ApsaraDB RDS for MySQL) dan tujuan (kluster Elasticsearch). Klik Test Connectivity and Proceed.
2. Pilih objek yang akan disinkronkan.
3. Konfigurasikan pengaturan advanced. Tutorial ini menggunakan nilai default.
4. Pada sublangkah Data Verification, pilih Apply _routing Policy to No Tables.
  
  Pengaturan ini diperlukan ketika kluster tujuan menjalankan Elasticsearch V7.X.
Simpan tugas, jalankan pemeriksaan awal (pre-check), dan beli instans DTS untuk memulai sinkronisasi. Setelah instans DTS dibeli, tugas akan dimulai secara otomatis. Pantau progres di halaman Sinkronisasi Data. Setelah sinkronisasi penuh selesai, lima baris awal tersedia di Elasticsearch.

Langkah 3: (Opsional) Verifikasi hasil sinkronisasi

Masuk ke Konsol Kibana untuk kluster Elasticsearch Anda. Di pojok kiri atas, pilih > Management > Dev Tools, lalu klik tab Console.

Verifikasi sinkronisasi data penuh

Jalankan kueri berikut:

GET /es_test/_search

Tanggapan berikut mengonfirmasi bahwa semua lima baris telah disinkronkan:

{
  "took" : 10,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}

Verifikasi sinkronisasi data inkremental

Masukkan baris baru ke tabel sumber:

INSERT INTO `test_mysql`.`es_test` (`id`,`name`,`age`,`hobby`) VALUES (6,'user6',30,'dance');

Setelah baris tersebut direplikasi, jalankan kembali GET /es_test/_search. Tanggapan sekarang menampilkan 6 hits, termasuk catatan baru:

{
  "took" : 541,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 6,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "6",
        "_score" : 1.0,
        "_source" : {
          "name" : "user6",
          "id" : 6,
          "age" : 30,
          "hobby" : "dance"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}

Langkah selanjutnya

Ikhtisar skenario sinkronisasi data — kombinasi sumber dan tujuan yang didukung
Menghapus objek dari tugas sinkronisasi data
Menambahkan objek ke tugas sinkronisasi data
Pemetaan tipe data untuk sinkronisasi skema