Sinkronisasi data penuh dan inkremental dari MySQL ke Elasticsearch - DataWorks

Data Integration memungkinkan Anda menyinkronkan seluruh database dari sumber MySQL ke Elasticsearch secara real-time. Topik ini menggunakan MySQL sebagai sumber dan Elasticsearch sebagai tujuan untuk menunjukkan cara mengonfigurasi tugas sinkronisasi real-time yang menggabungkan Full Synchronization dengan Incremental Synchronization (change data capture atau CDC).

Prasyarat

Data sources
- Anda telah membuat sumber data MySQL dan Elasticsearch. Untuk informasi selengkapnya, lihat Configure a data source.
- Anda telah mengaktifkan binary logging untuk sumber data MySQL. Untuk informasi selengkapnya, lihat Prerequisites.
Resource group: Anda telah membeli Serverless resource group.
Network connectivity: Anda telah menetapkan koneksi jaringan antara resource group dan sumber data. Untuk informasi selengkapnya, lihat Overview of network connection solutions.

Konfigurasikan tugas

Langkah 1: Buat tugas sinkronisasi

Buka halaman Data Integration.
Login ke DataWorks console. Di bilah navigasi atas, pilih wilayah yang diinginkan. Di panel navigasi kiri, pilih Data Integration > Data Integration. Pada halaman yang muncul, pilih ruang kerja yang diinginkan dari daftar drop-down dan klik Go to Data Integration.
Di panel navigasi kiri, klik Synchronization Task. Di bagian atas halaman, klik Create Synchronization Task dan konfigurasikan informasi tugas:
- Source Type: Pilih MySQL sebagai sumber.
- Destination Type: Pilih Elasticsearch sebagai tujuan.
- Specific Type: Pilih Real-time synchronization for an entire database.
- Synchronization Mode:
  - Schema Migration: Secara otomatis membuat struktur indeks yang sesuai (seperti indeks dan pemetaan bidang) di tujuan, tetapi tidak menyertakan data.
  - Full Synchronization (Opsional): Menyalin semua data historis dari tabel sumber yang ditentukan ke tujuan.
  - Incremental Sync (Opsional): Setelah full synchronization selesai, terus-menerus menangkap dan menyinkronkan perubahan data (insert, update, dan delete) dari sumber ke tujuan.

Langkah 2: Konfigurasikan sumber data dan sumber daya komputasi

Untuk Source, pilih sumber data MySQL Anda. Untuk Destination, pilih sumber data Elasticsearch Anda.

Langkah 3: Konfigurasikan solusi sinkronisasi

2. Pemetaan indeks tujuan

Actions

Description

Refresh

Sistem secara otomatis mencantumkan tabel sumber yang Anda pilih, tetapi Anda harus merefresh pemetaan untuk mengonfirmasi atribut indeks tujuan.

Pilih tabel yang akan disinkronkan dan klik Batch Refresh Mapping.
Nama indeks tujuan: Nama indeks tujuan dibuat secara otomatis berdasarkan aturan Customize Mapping Rules for Destination Table Names. Nilai default-nya adalah ${source_database_name}_${table_name}. Jika indeks dengan nama tersebut belum ada di Tujuan, sistem akan membuatnya secara otomatis.

Custom Mapping Rule for Destination Index Name (Opsional)

Sistem memiliki aturan penamaan indeks default: ${source_database_name}_${table_name}. Anda juga dapat menambahkan aturan kustom dengan mengklik tombol Edit di kolom Customize Mapping Rules for Destination Table Names.

Nama aturan: Tentukan nama untuk aturan tersebut. Kami menyarankan menggunakan nama yang jelas mencerminkan tujuan bisnisnya.
Nama indeks tujuan: Susun nama indeks tujuan dengan mengklik ikon dan menggabungkan opsi dari Manual Input dan Built-in Variable. Variabel yang didukung meliputi nama sumber data sumber, nama database sumber, dan nama tabel sumber.
Edit variabel bawaan: Anda dapat menerapkan transformasi string pada variabel bawaan.

Fitur ini mendukung skenario berikut:

Tambahkan awalan atau akhiran: Tambahkan awalan atau akhiran ke nama tabel sumber dengan menetapkan konstanta.
Rule configuration
Result
Penggantian string terpadu: Ganti string dev_ dalam nama tabel sumber dengan prd_.
Rule configuration
Result
Tulis data dari beberapa tabel ke satu indeks: Tetapkan nama indeks tujuan sebagai konstanta.
Rule configuration
Result

Edit field type mapping (Opsional)

Sistem menyediakan pemetaan default antara Source type dan Destination type. Anda dapat mengklik Edit Mapping of Field Data Types di pojok kanan atas tabel untuk menyesuaikan pemetaan tipe bidang dari sumber ke tujuan. Setelah dikonfigurasi, klik Apply and Refresh Mapping.

Saat mengedit pemetaan tipe bidang, pastikan aturan konversinya benar. Aturan yang salah dapat menyebabkan kegagalan konversi tipe, menghasilkan Dirty Data, dan memengaruhi eksekusi tugas.

Edit destination index (Opsional)

Sistem menggunakan aturan pemetaan nama indeks untuk membuat indeks tujuan baru atau menggunakan kembali indeks yang sudah ada dengan nama yang sama.

DataWorks secara otomatis menghasilkan struktur indeks tujuan berdasarkan skema tabel sumber. Dalam kebanyakan kasus, tidak diperlukan intervensi manual.

Saat indeks tujuan memiliki status To be created, Anda dapat menambahkan bidang baru ke strukturnya. Lakukan langkah-langkah berikut:

Tambahkan bidang ke indeks tujuan
- Tambahkan bidang ke satu indeks: Klik ikon di kolom Destination Index Name dan edit Statement Used to Create Index untuk menambahkan bidang.
  - Dynamic Mapping Status: Mengontrol apakah bidang baru dari tabel sumber ditambahkan ke indeks tujuan selama sinkronisasi data. Nilai yang valid adalah:
    - true: Jika bidang baru terdeteksi di tabel sumber, bidang tersebut ditambahkan ke indeks tujuan dan dapat dicari. Ini adalah nilai default.
    - false: Jika bidang baru terdeteksi di tabel sumber, bidang tersebut ditambahkan ke indeks tujuan tetapi tidak dapat dicari.
    - strict: Jika bidang baru terdeteksi di tabel sumber, sinkronisasi bidang tersebut ditolak, dan kesalahan dilemparkan. Anda dapat melihat detail kesalahan di log.
    - runtime: Jika bidang baru terdeteksi di tabel sumber, bidang tersebut tidak ditambahkan ke pemetaan indeks tujuan. Sebagai gantinya, bidang tersebut diperlakukan sebagai runtime field saat kueri, sehingga dapat digunakan dalam perhitungan skrip dan pencarian.
    Untuk informasi selengkapnya tentang dynamic mapping, lihat dynamic.
  - Shards dan Replica Shards: Jumlah shard utama dan shard replika untuk indeks. Indeks dibagi menjadi beberapa shard dan didistribusikan ke berbagai node Elasticsearch untuk memungkinkan pencarian terdistribusi dan meningkatkan efisiensi kueri. Untuk informasi selengkapnya, lihat Basic concepts.
    Catatan
    Parameter Shards dan Replica Shards tidak dapat diubah setelah indeks dibuat. Nilai default-nya adalah 1.
- Tambahkan bidang secara massal: Pilih semua tabel yang akan disinkronkan, lalu di bagian bawah tabel, pilih Batch Modify > Destination Index Structure - Batch Add Fields.

Value assignment

Bidang sumber dipetakan secara otomatis jika namanya sesuai dengan bidang di Tujuan. Anda harus menetapkan nilai secara manual untuk bidang yang baru ditambahkan dan properti indeks tujuan dari langkah sebelumnya. Lakukan langkah-langkah berikut:

Tetapkan nilai untuk satu indeks: Klik tombol Configuration di kolom Value assignment untuk menetapkan nilai.
Tetapkan nilai secara massal: Di bagian bawah daftar, pilih Batch Modify > Value assignment untuk menetapkan nilai ke bidang yang sama di beberapa indeks tujuan.

Anda dapat menetapkan konstanta atau variabel. Alihkan tipe di menu Value Type. Opsi berikut didukung:

Destination index field:
- Manual assignment: Masukkan nilai konstanta, seperti abc.
- Source field: Gunakan nilai bidang tabel sumber sebagai nilai bidang tujuan. Anda dapat memilih nilai bidang atau nilai waktu.
  - Field value: Menulis nilai bidang sumber langsung ke tujuan.
  - Time value: Jika bidang sumber berisi nilai waktu, Anda dapat memprosesnya berdasarkan format yang berbeda dan menentukan Destination Format untuk memformat nilai yang diekstraksi.
    - Time string: String yang merepresentasikan waktu, seperti "2018-10-23 02:13:56" atau "2021/05/18". Uraikan string menjadi objek waktu dengan menentukan format waktu sumber dan tujuan. Untuk contoh di atas, Anda dapat menggunakan format seperti yyyy-MM-dd HH:mm:ss dan yyyy/MM/dd.
    - Time object: Pilih tipe ini jika nilai sumber sudah berupa tipe waktu, seperti Date atau Datetime.
    - Unix timestamp (seconds): Timestamp dalam satuan detik. Juga mendukung angka atau string dalam format timestamp 10 digit, seperti 1610529203 atau "1610529203".
    - Unix timestamp (milliseconds): Timestamp dalam satuan milidetik. Juga mendukung angka atau string dalam format timestamp 13 digit, seperti 1610529203002 atau "1610529203002".
- Select variable: Pilih variabel yang disediakan sistem sebagai sumber nilai.
- Function: Gunakan fungsi untuk melakukan transformasi sederhana pada bidang sumber sebelum menetapkan nilainya. Untuk detail penggunaan, lihat Use a function expression to assign a value to a destination field.
Tetapkan properti indeks tujuan: Tetapkan nilai untuk kunci utama indeks tujuan, yang digunakan untuk sinkronisasi. Anda dapat menggabungkan beberapa bidang sumber untuk membuat kunci utama komposit. Pastikan nilai yang dihasilkan unik.

Source split column

Di Source Split Column, Anda dapat memilih bidang dari tabel sumber atau memilih Not Split. Saat tugas sinkronisasi berjalan, tugas tersebut dibagi menjadi beberapa tugas berdasarkan bidang ini untuk membaca data secara konkuren dan dalam batch.

Gunakan kunci utama tabel sebagai source split column. Tipe string, float, dan date tidak didukung.

Saat ini, source split column hanya didukung untuk sumber MySQL.

Skip full synchronization

Jika Anda mengaktifkan Full Synchronization untuk tugas, Anda dapat menonaktifkannya secara selektif untuk tabel tertentu. Ini berguna jika data lengkap untuk tabel tersebut telah disinkronkan ke tujuan melalui cara lain.

Full condition

Terapkan filter pada data sumber selama fase full synchronization. Masukkan hanya konten klausa WHERE, tanpa kata kunci WHERE.

Configure DML Rule

Penanganan pesan DML memungkinkan Anda memfilter dan mengontrol data perubahan (Insert, Update, Delete) yang ditangkap dari Sumber sebelum ditulis ke Tujuan. Aturan ini hanya berlaku selama fase incremental synchronization.

Langkah 4: Konfigurasikan pengaturan lanjutan

Konfigurasikan parameter lanjutan

Untuk menyesuaikan tugas dan memenuhi persyaratan sinkronisasi kustom, buka tab Advanced Parameters untuk memodifikasi parameter lanjutan.

Di pojok kanan atas halaman, klik Advanced Settings untuk membuka halaman konfigurasi parameter lanjutan.
Ubah nilai parameter sesuai dengan deskripsi yang diberikan.
Anda juga dapat menggunakan konfigurasi berbasis AI. Masukkan perintah dalam bahasa alami, seperti menyesuaikan konkurensi tugas, dan model bahasa besar akan menghasilkan rekomendasi nilai parameter. Anda kemudian dapat memilih apakah akan menerima parameter yang dihasilkan AI tersebut.

Penting

Ubah parameter ini hanya jika Anda sepenuhnya memahami tujuannya. Pengaturan yang salah dapat menyebabkan penundaan tugas, konsumsi sumber daya berlebihan yang menghambat tugas lain, kehilangan data, atau masalah tak terduga lainnya.

Langkah 7: Konfigurasi lainnya

Konfigurasi alarm

1. Tambahkan Alarm

(1) Klik Create Rule untuk mengonfigurasi aturan alarm.

Tetapkan Alert Reason untuk memantau metrik seperti Business delay, Failover, Task status, DDL Notification, dan Task Resource Utilization untuk tugas tersebut. Anda dapat menetapkan level alarm CRITICAL atau WARNING berdasarkan ambang batas yang ditentukan.

Dengan menetapkan Configure Advanced Parameters, Anda dapat mengontrol interval waktu pengiriman pesan alarm untuk mencegah kelelahan akibat notifikasi berlebihan dan penumpukan pesan.
Jika Anda memilih Business delay, Task status, atau Task Resource Utilization sebagai alasan alarm, Anda juga dapat mengaktifkan notifikasi pemulihan untuk memberi tahu penerima ketika tugas kembali normal.

(2) Kelola aturan alarm.

Untuk aturan alarm yang telah dibuat, Anda dapat menggunakan sakelar alarm untuk mengontrol apakah aturan alarm diaktifkan. Kirim alarm ke penerima tertentu berdasarkan level alarm.

2. Lihat Alarm

Bentangkan More > Configure Alert Rule di daftar tugas untuk membuka halaman event alarm dan melihat informasi alarm yang telah terjadi.

Konfigurasi resource group

Anda dapat mengelola resource group yang digunakan oleh tugas dan konfigurasinya di panel Configure Resource Group di pojok kanan atas antarmuka.

1. Lihat dan ganti resource group

Klik Configure Resource Group untuk melihat resource group yang saat ini terikat pada tugas.
Untuk mengganti resource group, alihkan ke resource group lain yang tersedia di sini.

2. Sesuaikan sumber daya dan troubleshooting error "insufficient resources"

Jika log tugas menampilkan pesan seperti Please confirm whether there are enough resources..., unit komputasi (CUs) yang tersedia dari resource group saat ini tidak cukup untuk memulai atau menjalankan tugas. Anda dapat menambah jumlah CUs yang digunakan oleh tugas di panel Configure Resource Group untuk mengalokasikan lebih banyak sumber daya komputasi.

Untuk rekomendasi pengaturan sumber daya, lihat Data Integration Recommended CUs. Sesuaikan pengaturan berdasarkan kondisi aktual.

Konfigurasi parameter lanjutan

Untuk persyaratan sinkronisasi kustom, klik Configure di kolom Advanced Settings untuk memodifikasi parameter lanjutan.

Klik Advanced Settings di pojok kanan atas antarmuka untuk membuka halaman konfigurasi parameter lanjutan.
Ubah nilai parameter sesuai petunjuk. Arti setiap parameter dijelaskan setelah nama parameter.

Penting

Pahami parameter sepenuhnya sebelum memodifikasi untuk mencegah masalah seperti penundaan tugas, konsumsi sumber daya berlebihan yang menghambat tugas lain, atau kehilangan data.

Kelola tugas

Edit tugas

Di halaman Data Integration > Synchronization Task, temukan tugas sinkronisasi yang telah dibuat. Di kolom Operation, klik More, lalu klik Edit untuk memodifikasi informasi tugas. Langkah-langkahnya sama seperti saat mengonfigurasi tugas baru.
Untuk tugas yang tidak berjalan, Anda dapat langsung memodifikasi konfigurasi, menyimpannya, dan menerbitkan tugas ke lingkungan produksi untuk menerapkan perubahan.
Untuk tugas yang Running, jika Anda mengedit dan menerbitkan tugas tanpa memilih Start immediately after deployment, tombol aksi berubah menjadi Apply Updates. Anda harus mengklik tombol ini agar perubahan berlaku di lingkungan produksi.
Saat Anda mengklik Apply Update, sistem melakukan tindakan berikut secara berurutan: Stop, Publish, dan Restart.
- Jika perubahan melibatkan menambahkan tabel atau mengganti tabel yang ada:
  Anda tidak dapat memilih Checkpoint saat menerapkan pembaruan. Setelah Anda mengonfirmasi, sistem melakukan Schema Migration dan Full Data Initialization untuk tabel baru. Setelah inisialisasi penuh selesai, sistem mulai melakukan sinkronisasi inkremental bersama tabel-tabel asli lainnya.
- Jika Anda memodifikasi informasi lainnya:
  Anda dapat memilih Checkpoint saat menerapkan pembaruan. Setelah Anda mengonfirmasi, tugas dilanjutkan dari Checkpoint yang ditentukan. Jika Anda tidak menentukan Checkpoint, tugas dilanjutkan dari Checkpoint terakhir yang terekam.
Tabel yang tidak diubah tidak terpengaruh. Setelah pembaruan dan restart, tabel tersebut akan dilanjutkan dari Checkpoint terakhir yang terekam.

Lihat tugas

Setelah membuat tugas sinkronisasi, Anda dapat melihat daftar semua tugas yang telah dibuat beserta informasi dasarnya di halaman Synchronization Task.

Di kolom Actions, Anda dapat Start atau Stop tugas sinkronisasi. Di bawah More, Anda dapat melakukan operasi lain seperti Edit dan View.
Untuk tugas yang sedang berjalan, Anda dapat melihat status dasarnya di Execution Overview. Klik area tertentu dalam ikhtisar untuk melihat informasi eksekusi terperinci.
Tugas sinkronisasi database real-time dari MySQL ke Elasticsearch terdiri dari tiga langkah:
- Schema Migration: Termasuk metode pembuatan indeks tujuan (indeks yang ada/indeks yang dibuat otomatis). Jika indeks dibuat otomatis, pernyataan DDL ditampilkan.
- Full Data Initialization: Termasuk informasi tentang tabel yang disinkronkan, progresnya, dan jumlah baris yang ditulis.
- Real-time Synchronization: Termasuk statistik sinkronisasi real-time, seperti progres real-time, catatan DDL, catatan DML, dan informasi peringatan.

Jalankan ulang tugas sinkronisasi

Dalam beberapa kasus khusus, jika Anda menambahkan atau menghapus tabel dari sumber, atau mengubah skema atau nama tabel tujuan, Anda dapat mengklik More di kolom Operation tugas sinkronisasi lalu mengklik Rerun untuk menjalankan ulang tugas setelah perubahan. Selama proses rerun, tugas sinkronisasi hanya menyinkronkan data dari tabel yang baru ditambahkan ke tujuan atau hanya dari tabel sumber yang dipetakan ke tabel tujuan yang skema atau namanya diubah.

Jika Anda ingin menjalankan ulang tugas sinkronisasi tanpa mengubah konfigurasi tugas, klik More di kolom Actions lalu klik Rerun untuk menjalankan ulang tugas guna melakukan full synchronization dan incremental synchronization lagi.
Jika Anda ingin menjalankan ulang tugas sinkronisasi setelah menambahkan atau menghapus tabel dari tugas, klik Complete setelah perubahan. Dalam hal ini, Apply Updates ditampilkan di kolom Actions tugas sinkronisasi. Klik Apply Updates untuk memicu sistem menjalankan ulang tugas sinkronisasi. Selama proses rerun, tugas sinkronisasi menyinkronkan data dari tabel yang baru ditambahkan ke tujuan. Data di tabel asli tidak disinkronkan lagi.

Breakpoint resume

Kasus penggunaan

Menyetel ulang Checkpoint secara manual saat memulai atau me-restart tugas berguna dalam skenario berikut:

Pemulihan tugas dan kelanjutan data: Jika tugas terganggu, Anda mungkin perlu menentukan waktu gangguan sebagai Checkpoint awal baru untuk memastikan data dilanjutkan secara akurat dari titik tersebut.
Troubleshooting dan rollback data: Jika Anda menemukan data yang hilang atau tidak normal setelah sinkronisasi, Anda dapat melakukan rollback Checkpoint ke waktu sebelum masalah terjadi untuk memproses ulang dan memperbaiki data yang bermasalah.
Perubahan konfigurasi tugas besar-besaran: Setelah melakukan perubahan signifikan pada konfigurasi tugas, seperti memodifikasi struktur indeks tujuan atau pemetaan bidang, kami menyarankan untuk menyetel ulang Checkpoint ke waktu tertentu untuk memastikan akurasi data di bawah konfigurasi baru.

Petunjuk

Ketika Anda mengklik Start, pilih opsi Whether to reset the site di jendela pop-up:

Jangan reset Checkpoint: Tugas dilanjutkan dari waktu berhenti terakhir (Checkpoint terakhir).
Reset Checkpoint dan pilih waktu: Tugas dimulai dari waktu yang ditentukan. Pastikan waktu yang dipilih tidak lebih awal dari waktu tersedia paling awal di binary log (binlog) sumber.

Penting

Jika Anda mengalami kesalahan yang menunjukkan bahwa Checkpoint tidak valid atau tidak ada, coba solusi berikut:

Reset Checkpoint: Saat memulai tugas sinkronisasi real-time, reset Checkpoint dan pilih waktu tersedia paling awal dari database sumber.
Sesuaikan periode retensi log: Jika Checkpoint database telah kedaluwarsa, pertimbangkan untuk memperpanjang periode retensi log di pengaturan database, misalnya menjadi 7 hari.
Sinkronisasi ulang data: Jika data telah hilang, pertimbangkan untuk melakukan full synchronization baru atau konfigurasikan tugas sinkronisasi offline untuk menyinkronkan manual data yang hilang.

FAQ

Untuk pertanyaan umum tentang sinkronisasi database real-time, lihat Data Integration FAQ dan Data Integration errors.