Sinkronisasi data dari PolarDB for MySQL ke Elasticsearch menggunakan DTS - Elasticsearch

Jika kueri lambat pada PolarDB for MySQL memengaruhi performa aplikasi, Anda dapat menggunakan Data Transmission Service (DTS) untuk mereplikasi data produksi ke kluster Alibaba Cloud Elasticsearch secara real time. Setelah data berada di Elasticsearch, Anda dapat menjalankan pencarian teks lengkap dan analitik yang cepat tanpa membebani database sumber.

Cara kerja

DTS menangkap perubahan tingkat baris dari binary log (binlog) PolarDB for MySQL dan mengalirkannya ke Elasticsearch. Sinkronisasi berlangsung dalam dua fase berurutan:

Phase	Trigger	Apa yang dilakukan DTS
Full data sync	Task start	Membaca snapshot dari tabel sumber dan mengindeks semua baris yang ada ke Elasticsearch
Incremental data sync	Setelah full sync selesai	Memutar ulang operasi INSERT, DELETE, dan UPDATE dari binlog secara real time

Karena DTS mengandalkan binlog untuk sinkronisasi inkremental, binary logging harus diaktifkan pada kluster PolarDB sebelum tugas dimulai. DTS tidak menangkap operasi DDL (seperti DROP COLUMN atau RENAME TABLE)—hanya operasi DML (INSERT, DELETE, UPDATE). Perbedaan ini penting: jika Anda menjalankan operasi DDL pada tabel sumber saat sinkronisasi sedang berjalan, indeks Elasticsearch akan tidak selaras dan memerlukan pemulihan manual. Lihat Batasan untuk langkah pemulihan.

Prasyarat

Sebelum memulai, pastikan Anda memiliki:

Kluster PolarDB for MySQL 8.0.1 Enterprise Edition dengan binary logging yang diaktifkan
Kluster Alibaba Cloud Elasticsearch V7.10 dengan Auto Indexing yang diaktifkan
Akses ke konsol DTS baru

Batasan

Limitation	Impact	Recovery
Operasi DDL tidak direplikasi	Jika Anda menjalankan pernyataan DDL (seperti menghapus atau mengganti nama kolom) pada tabel sumber saat tugas sedang berjalan, indeks Elasticsearch menjadi tidak selaras. Hal ini dapat menyebabkan kehilangan data atau error kueri di tujuan.	Hapus tabel tersebut dari tugas sinkronisasi, hapus indeks yang sesuai dari Elasticsearch, lalu tambahkan kembali tabel tersebut. Lihat Remove an object from a data synchronization task dan Add an object to a data synchronization task.
Menambahkan kolom memerlukan langkah tambahan	Menambahkan kolom merupakan operasi DDL. Jika Anda menjalankan ALTER TABLE tanpa memperbarui pemetaan Elasticsearch terlebih dahulu, bidang baru tidak akan muncul di indeks, sehingga menyebabkan data tidak lengkap di tujuan.	Perbarui pemetaan indeks di Elasticsearch untuk menyertakan bidang baru terlebih dahulu. Kemudian jalankan ALTER TABLE pada sumber. Setelah DDL selesai, jeda tugas sinkronisasi dan mulai ulang.
Full sync meningkatkan beban pada sumber dan tujuan	DTS membaca dari PolarDB dan menulis ke Elasticsearch secara bersamaan selama full sync awal. Hal ini dapat menurunkan performa aplikasi di kedua sisi jika sumber daya sudah terbebani.	Jalankan full sync selama jam sepi ketika utilisasi CPU baik di sumber maupun tujuan berada di bawah 30%. Jika full sync gagal selama jam sibuk, mulai ulang tugas tersebut. Sinkronisasi inkremental selama jam sibuk dapat menyebabkan keterlambatan replikasi tetapi tidak akan gagal.

Menyiapkan dan menjalankan sinkronisasi

Contoh ini menggunakan kluster PolarDB for MySQL 8.0.1 Enterprise Edition sebagai sumber dan kluster Alibaba Cloud Elasticsearch V7.10 sebagai tujuan.

Langkah 1: Siapkan database sumber

Buat kluster PolarDB for MySQL 8.0.1 Enterprise Edition. Untuk detailnya, lihat Purchase an Enterprise Edition cluster.
Aktifkan binary logging pada kluster. Binary logging diperlukan untuk sinkronisasi inkremental—DTS membaca binlog untuk menangkap perubahan INSERT, DELETE, dan UPDATE secara real time. Untuk detailnya, lihat Enable binary logging.
Buat akun database dan database bernama test_polardb. Untuk detailnya, lihat Create and manage a database account dan Database management operations.

Pada database test_polardb, buat tabel bernama product:

CREATE TABLE `product` (
    `id` bigint(32) NOT NULL AUTO_INCREMENT,
    `name` varchar(32) NULL,
    `price` varchar(32) NULL,
    `code` varchar(32) NULL,
    `color` varchar(32) NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB
DEFAULT CHARACTER SET=utf8;

Masukkan data uji ke dalam tabel:

INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (1,'mobile phone A','2000','amp','golden');
INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (2,'mobile phone B','2200','bmp','white');
INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (3,'mobile phone C','2600','cmp','black');
INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (4,'mobile phone D','2700','dmp','red');
INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (5,'mobile phone E','2800','emp','silvery');

Langkah 2: Siapkan kluster Elasticsearch tujuan

Buat kluster Alibaba Cloud Elasticsearch V7.10. Untuk detailnya, lihat Create an Alibaba Cloud Elasticsearch cluster.
Aktifkan Auto Indexing pada kluster. DTS membuat indeks Elasticsearch secara otomatis saat menulis data—Auto Indexing harus diaktifkan agar fitur ini berfungsi. Untuk detailnya, lihat Configure the YML file.

Langkah 3: Buat tugas sinkronisasi DTS

Buka halaman Data Synchronization di konsol DTS baru.
Klik Create Task.

Konfigurasikan tugas tersebut. Untuk deskripsi lengkap setiap parameter, lihat Synchronize data from a PolarDB for MySQL cluster.

Konfigurasikan sumber (kluster PolarDB for MySQL) dan tujuan (kluster Elasticsearch). Parameter utama:

Parameter	Tempat pengaturan	Nilai yang dimasukkan
Source type	Source configuration	PolarDB for MySQL
Destination type	Destination configuration	Alibaba Cloud Elasticsearch
Cluster endpoint and credentials	Source and destination sections	Detail koneksi untuk kluster PolarDB dan kluster Elasticsearch Anda

Klik Test Connectivity and Proceed untuk memvalidasi koneksi.

Pilih objek yang akan disinkronkan—pilih database test_polardb dan tabel product.
Pertahankan pengaturan advanced default.
Pada sublangkah Data Verification, pilih Apply _routing Policy to No Tables. > Note: Untuk kluster Elasticsearch V7.X, Anda harus memilih Apply _routing Policy to No Tables.

Simpan tugas, selesaikan pemeriksaan awal, dan beli instans DTS untuk memulai sinkronisasi. Setelah instans DTS dibeli, full data sync akan dimulai secara otomatis. Pantau progres di halaman Data Synchronization. Setelah full sync selesai, DTS beralih ke incremental sync dan mulai mengalirkan perubahan binlog ke Elasticsearch secara real time.

Verifikasi hasil sinkronisasi (opsional)

Gunakan konsol Kibana untuk memastikan bahwa data dari PolarDB telah direplikasi dengan benar.

Login ke konsol Kibana kluster Elasticsearch. Untuk detailnya, lihat Log on to the Kibana console.
Di pojok kiri atas, pilih > Management > Dev Tools, lalu klik tab Console.

Verifikasi hasil full sync. Jalankan:

GET /product/_search

Tanggapan harus menampilkan semua 5 catatan yang dimasukkan pada Langkah 1:

{
  "took" : 3,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "mobile phone C",
          "price" : "2600",
          "code" : "cmp",
          "color" : "black"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "mobile phone E",
          "price" : "2800",
          "code" : "emp",
          "color" : "silvery"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "mobile phone D",
          "price" : "2700",
          "code" : "dmp",
          "color" : "red"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "mobile phone B",
          "price" : "2200",
          "code" : "bmp",
          "color" : "white"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "mobile phone A",
          "price" : "2000",
          "code" : "amp",
          "color" : "golden"
        }
      }
    ]
  }
}

Verifikasi incremental sync. Masukkan catatan baru ke tabel sumber:

INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (6,'mobile phone F','2750','fmp','white');

Setelah catatan direplikasi, jalankan kembali GET /product/_search. Tanggapan sekarang harus menampilkan 6 hits, termasuk catatan baru:

{
  "took" : 439,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 6,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "mobile phone C",
          "price" : "2600",
          "code" : "cmp",
          "color" : "black"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "mobile phone E",
          "price" : "2800",
          "code" : "emp",
          "color" : "silvery"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "mobile phone D",
          "price" : "2700",
          "code" : "dmp",
          "color" : "red"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "mobile phone B",
          "price" : "2200",
          "code" : "bmp",
          "color" : "white"
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "6",
        "_score" : 1.0,
        "_source" : {
          "code" : "fmp",
          "color" : "white",
          "price" : "2750",
          "name" : "mobile phone F",
          "id" : 6
        }
      },
      {
        "_index" : "product",
        "_type" : "product",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "mobile phone A",
          "price" : "2000",
          "code" : "amp",
          "color" : "golden"
        }
      }
    ]
  }
}

Langkah berikutnya

Untuk mencari dan menganalisis data di kluster Elasticsearch Anda, lihat What is Alibaba Cloud Elasticsearch?
Untuk menghapus objek dari tugas sinkronisasi yang sedang berjalan, lihat Remove an object from a data synchronization task.
Untuk menambahkan kembali objek ke tugas sinkronisasi yang sedang berjalan, lihat Add an object to a data synchronization task.