Contoh skenario untuk mengonfigurasi tugas sinkronisasi inkremental - DataWorks

Anda dapat menggunakan kondisi filter dalam node sinkronisasi batch untuk menyinkronkan data penuh atau data inkremental. Dengan kondisi filter tersebut, Data Integration hanya menyinkronkan data yang memenuhi kriteria tertentu. Anda juga dapat menggabungkan parameter penjadwalan dengan kondisi filter untuk secara dinamis memfilter data berdasarkan waktu proses node, sehingga memungkinkan sinkronisasi inkremental. Topik ini menjelaskan cara mengonfigurasi node sinkronisasi batch untuk sinkronisasi inkremental.

Catatan penggunaan

Sinkronisasi inkremental tidak didukung untuk beberapa sumber data, seperti Hbase dan Tablestore (OTSStream). Untuk memeriksa apakah sinkronisasi inkremental didukung untuk sumber data tertentu, lihat dokumentasi plugin reader yang sesuai.

Parameter yang diperlukan untuk sinkronisasi inkremental bervariasi tergantung pada plugin reader. Untuk detailnya, lihat dokumentasi plugin spesifik dan Sumber data dan plugin yang didukung. Contohnya:

Reader plug-in	Required parameter	Supported syntax
MySQL Reader	where Catatan Dalam mode wizard, ini adalah parameter kondisi filter.	Database syntax Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari rentang waktu tertentu setiap hari.
MongoDB Reader	query Catatan Dalam mode wizard, ini adalah parameter Search Condition.	Similar to database syntax Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari rentang waktu tertentu setiap hari.
OSS Reader	Object	Specify the object path Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari file tertentu setiap hari.
...	...	...

Konfigurasikan sinkronisasi inkremental

Dalam node sinkronisasi batch Data Integration, Anda dapat menggunakan parameter penjadwalan untuk menentukan path dan rentang data untuk tabel sumber dan tujuan. Konfigurasinya sama seperti jenis node lainnya.

Saat waktu proses, sistem mengganti semua placeholder parameter yang dikonfigurasi dalam node dengan nilai aktual yang diwakili oleh ekspresi parameter penjadwalan, lalu melakukan sinkronisasi data.

Ambil contoh sinkronisasi data MySQL:

Jika Anda tidak mengonfigurasi Data Filtering, seluruh data disinkronkan ke tabel tujuan secara default.
Jika Anda mengonfigurasi Data Filtering, hanya data yang memenuhi kondisi filter yang disinkronkan ke tabel tujuan.

Nama partisi tabel MaxCompute tujuan ditentukan oleh parameter penjadwalan. $bizdate merepresentasikan tanggal bisnis. Saat tugas terjadwal dijalankan, ekspresi partisi yang dikonfigurasi untuk tugas tersebut diganti dengan tanggal bisnis yang diwakili oleh parameter penjadwalan. Untuk instruksi konfigurasi detail mengenai ekspresi parameter penjadwalan, lihat Skema penggunaan parameter penjadwalan dalam Data Integration. Ambil contoh tugas sinkronisasi batch. Anda perlu mengonfigurasi parameter bizdate di tiga tempat untuk menerapkan sinkronisasi inkremental: Di bagian Data Filtering sumber, masukkan STR_TO_DATE('${bizdate}','%Y%m%d') <= gmt_modify_time AND gmt_modify_time < DATE_ADD(STR_TO_DATE('${bizdate}','%Y%m%d'), interval 1 day) untuk memfilter data yang dimodifikasi pada tanggal bisnis. Di bagian Partition Information tujuan, masukkan pt=${bizdate} untuk menulis data ke partisi tanggal yang sesuai, dan atur Cleanup Rule menjadi Clean up existing data before writing (Insert Overwrite). Di bagian Parameters dalam Schedule Settings di sisi kanan, masukkan bizdate=$bizdate agar sistem penjadwalan secara otomatis mengganti ${bizdate} dengan tanggal bisnis aktual saat waktu proses. Saat Anda mengonfigurasi sinkronisasi data inkremental:

Sinkronisasi inkremental berdasarkan kolom bertipe waktu: Anda dapat menggunakan parameter penjadwalan untuk mengganti data bertipe waktu secara dinamis. Selama penjadwalan tugas, parameter penjadwalan secara otomatis diganti dengan nilai spesifik berdasarkan tanggal bisnis. Untuk informasi lebih lanjut tentang parameter penjadwalan, lihat Konfigurasikan parameter penjadwalan.
Sinkronisasi inkremental berdasarkan kolom non-bertipe waktu: Anda dapat menggunakan node assignment untuk mengonversi kolom ke tipe data target, lalu meneruskannya ke Data Integration untuk disinkronkan. Untuk informasi lebih lanjut tentang node assignment, lihat Buat node assignment.

Catatan

Saat mengonfigurasi tugas sinkronisasi inkremental, perhatikan hal-hal berikut:

Keamanan Clean up existing data before writing (Insert Overwrite): Ketika beberapa tugas sinkronisasi menulis ke partisi berbeda dari tabel MaxCompute yang sama, strategi Insert Overwrite aman. Strategi ini hanya membersihkan data partisi yang ditentukan oleh tugas saat ini dan tidak memengaruhi data di partisi lain tabel tersebut, sehingga mencegah konflik data atau penghapusan tidak disengaja.
Batasan overwrite batch rentang partisi: DataWorks tidak mendukung penentuan rentang jam (seperti hh=00-23) dalam konfigurasi partisi untuk overwrite batch. Untuk overwrite data beberapa jam, konfigurasikan tugas terpisah untuk setiap jam. Parameter partisi saat ini hanya mendukung satu nilai spesifik atau wildcard *.
Sintaks wildcard: Jika sumber berisi partisi tingkat jam tetapi tujuan hanya memiliki partisi tingkat hari, masukkan wildcard * di bidang partisi jam untuk mencocokkan seluruh data per jam. Masukkan * secara langsung tanpa tanda kutip (seperti "*"). Jika tidak, akan terjadi kesalahan sintaks.

Sinkronisasi inkremental terjadwal frekuensi tinggi berbasis timestamp

DataWorks mendukung sinkronisasi inkremental terjadwal berbasis timestamp dengan menggabungkan tugas sinkronisasi batch dan penjadwalan periodik (misalnya setiap 5 menit atau setiap jam). Pendekatan ini cocok untuk skenario sinkronisasi T+1 atau near-real-time dari RDS MySQL ke tujuan seperti SelectDB dan StarRocks. Sinkronisasi inkremental diimplementasikan melalui pemfilteran SQL tanpa memerlukan tugas CDC real-time, sehingga menghindari biaya tugas yang berjalan terus-menerus. Poin konfigurasi utama:

Dalam kondisi where sumber Data Filtering, gunakan kolom timestamp sebagai variabel untuk pemfilteran. Contohnya: gmt_modify_time >= '$[yyyymmddhhmiss-10/mi]' AND gmt_modify_time < '$[yyyymmddhhmiss]'.
Konfigurasikan parameter penjadwalan periodik (seperti $[yyyymmddhhmiss]) untuk menghitung rentang waktu secara dinamis, memastikan setiap eksekusi penjadwalan hanya menyinkronkan data inkremental dalam interval waktu yang ditentukan. Untuk konfigurasi detail parameter penjadwalan, lihat Skema penggunaan parameter penjadwalan dalam Data Integration.
Dalam pemetaan kolom tujuan, tambahkan secara manual parameter konstan yang dipetakan ke kolom partisi untuk mengaktifkan penulisan partisi dinamis.

Konfigurasi inkremental untuk tugas sinkronisasi batch tingkat database

Selain tugas sinkronisasi tabel tunggal, Anda juga dapat membuat tugas sinkronisasi batch tingkat database untuk menerapkan sinkronisasi inkremental periodik. Saat membuat tugas, pilih sinkronisasi inkremental dan konfigurasi kondisi inkremental dalam tugas sinkronisasi tingkat database. Hal ini memungkinkan sinkronisasi partisi inkremental tingkat hari yang efisien (misalnya, memfilter berdasarkan kolom create_time).

Pendekatan ini cocok untuk skenario di mana Anda ingin mengelola sinkronisasi beberapa tabel secara terpusat tetapi hanya memerlukan pemrosesan inkremental untuk tabel tertentu. Untuk proses konfigurasi lengkap tugas sinkronisasi batch tingkat database, lihat Konfigurasikan tugas sinkronisasi batch tingkat database.

Contoh

Menyinkronkan data historis: Jika Anda perlu menyinkronkan data inkremental historis ke partisi waktu yang sesuai di tabel tujuan, Anda dapat menggunakan fitur pengisian ulang data di Operation Center. Untuk informasi lebih lanjut tentang fitur pengisian ulang data, lihat Pengisian ulang data. Dalam konfigurasi node sinkronisasi data, pilih MySQL sebagai sumber data dan MaxCompute (ODPS) sebagai tujuan, dan atur nama tabel ke nilai seperti czd. Dalam kondisi filter data, gunakan ${bizdate} untuk mengontrol rentang inkremental (misalnya, STR_TO_DATE('${bizdate}','%Y%m%d') <= gmt_modify_time). Atur informasi partisi ke ds=${bizdate}, dan atur aturan pembersihan ke Clean up existing data before writing (Insert Overwrite). Di bagian parameter pengaturan jadwal, definisikan bizdate=$bizdate. Parameter penjadwalan ini secara otomatis diganti dengan nilai tanggal spesifik berdasarkan tanggal bisnis selama pengisian ulang. Saat menjalankan pengisian ulang, Anda dapat mengatur beberapa rentang tanggal bisnis (misalnya, 2022-05-01 hingga 2022-05-31 dan 2022-04-01 hingga 2022-04-30), pilih Immediately Run Backfill Instances Whose Scheduled Time Is Later Than the Current Time, dan pilih Ascending Order of Business Dates untuk eksekusi.
Sinkronisasi data inkremental dari ApsaraDB RDS ke MaxCompute

FAQ

Apa yang harus saya lakukan jika terjadi error partition not found setelah menggunakan parameter penjadwalan dalam filter partisi MaxCompute Reader?

Penyebab: Parameter penjadwalan yang dikonfigurasi tidak berhasil di-resolve ke nilai partisi aktual saat waktu proses, atau nilai yang di-resolve tidak sesuai dengan partisi aktual di tabel sumber.

Solusi: Jika nilai partisi diteruskan dari parameter outputs node hulu, periksa konfigurasi parameter di Data Studio dan pastikan kondisi berikut terpenuhi:

Nama parameter yang dikonfigurasi konsisten dengan nama parameter input.
Nilai parameter yang diteruskan persis sesuai dengan partisi aktual di MaxCompute.

Apakah tugas sinkronisasi batch DataWorks melakukan sinkronisasi penuh atau inkremental secara default? Bagaimana cara mengonfigurasi sinkronisasi inkremental untuk tabel sumber tanpa kolom partisi?

Perilaku default

Tugas sinkronisasi batch Data Integration DataWorks melakukan sinkronisasi penuh secara default, artinya seluruh data disinkronkan setiap kali. Sinkronisasi inkremental hanya diaktifkan ketika Anda mengonfigurasi kondisi Data Filtering yang dikombinasikan dengan parameter penjadwalan.

Menangani tabel sumber tanpa kolom partisi

Jika tabel database sumber (seperti tabel RDS) tidak memiliki kolom waktu atau partisi, Anda tidak dapat langsung memfilter data inkremental menggunakan kondisi where. Kami menyarankan Anda menambahkan kolom waktu (seperti dt atau gmt_modify_time) ke tabel sumber sebagai dasar untuk pemfilteran inkremental. Setelah kolom siap, konfigurasikan logika sinkronisasi inkremental dengan merujuk pada bagian "Konfigurasikan sinkronisasi inkremental" dalam topik ini.

Konsep kunci

Definisi parameter

splitPk (split key): Menentukan kolom kunci primer. DataWorks membagi data menjadi beberapa chunk berdasarkan rentang nilai kolom ini untuk memungkinkan pembacaan konkuren multi-threaded.
splitFactor (split factor): Mengontrol granularitas pemisahan. Nilai yang lebih besar menghasilkan pemisahan lebih halus dan lebih banyak thread baca.

Dampak kinerja dan rekomendasi

Mengaktifkan splitPk dan splitFactor meningkatkan beban pada database sumber. Untuk mengurangi beban pada sumber, kami merekomendasikan:

Mengurangi konkurensi, atau mengatur konkurensi tugas sinkronisasi batch tingkat database menjadi 1.
Pastikan kolom split (splitPk) diindeks untuk meningkatkan efisiensi baca.

Bagaimana cara memilih solusi sinkronisasi yang sesuai untuk bisnis saya?

Solusi sinkronisasi ditentukan oleh dua dimensi independen: mode eksekusi (sinkronisasi batch terjadwal atau sinkronisasi real-time) dan rentang data (penuh atau inkremental). Kedua dimensi ini dapat dikombinasikan secara bebas. Skenario yang dijelaskan dalam topik ini adalah sinkronisasi batch terjadwal + sinkronisasi inkremental.

Memilih mode eksekusi

Sinkronisasi batch terjadwal direkomendasikan secara default: sumber daya dilepas segera setelah tugas selesai, dan Anda dikenai biaya berdasarkan durasi eksekusi aktual. Pilih sinkronisasi real-time hanya dalam dua kasus berikut:

Bisnis Anda memerlukan latensi tingkat detik.
Anda harus menyinkronkan operasi DELETE fisik dari sumber.

Sinkronisasi real-time diperlukan dalam kasus-kasus ini karena sinkronisasi batch inkremental memfilter data menggunakan kondisi seperti timestamp. Catatan yang dihapus secara fisik di sumber tidak lagi muncul dalam hasil kueri dan karenanya tidak dihapus di tujuan. Sinkronisasi real-time hanya dirujuk di sini; konfigurasinya tidak dibahas dalam topik ini. Untuk informasi lebih lanjut, lihat Kemampuan sinkronisasi real-time tabel tunggal.

Memilih rentang data

Sinkronisasi penuh cocok untuk tabel sumber dengan volume data kecil, atau untuk skenario di mana data historis dimodifikasi di tempat dan tidak ada kolom waktu perubahan, sehingga rentang perubahan tidak dapat ditentukan.
Sinkronisasi inkremental memerlukan tabel sumber memiliki kolom waktu perubahan yang andal (seperti gmt_modify_time) atau primary key auto-increment. Ini cocok untuk volume data besar dengan rasio perubahan harian kecil, dan untuk skenario di mana Anda perlu menyimpan snapshot historis dalam partisi tanggal (tabel zip, juga dikenal sebagai tabel dimensi yang berubah lambat).
Jika tabel sumber tidak memiliki kolom semacam itu, lihat FAQ berikut dalam topik ini: Apakah tugas sinkronisasi batch DataWorks melakukan sinkronisasi penuh atau inkremental secara default? Bagaimana cara mengonfigurasi sinkronisasi inkremental untuk tabel sumber tanpa kolom partisi?.

Pemuatan awal dan volume data besar

Gunakan strategi "penuh dulu, lalu inkremental" untuk data historis, dan tulis kedua bagian tersebut ke partisi berbeda.
Untuk mengisi ulang partisi historis berdasarkan timestamp data aslinya, gunakan fitur pengisian ulang data. Untuk informasi lebih lanjut, lihat Pengelolaan instance pengisian ulang data.
Untuk volume data tingkat TB, tingkatkan konkurensi tugas dan sesuaikan CUs kelompok sumber daya Data Integration untuk menjaga efisiensi sinkronisasi. Untuk informasi lebih lanjut, lihat Buat dan gunakan kelompok sumber daya Serverless. Perhatikan bahwa konkurensi yang lebih tinggi meningkatkan beban pada database sumber. Untuk informasi lebih lanjut, lihat FAQ berikut dalam topik ini: Konsep kunci.