Konfigurasikan tugas sinkronisasi data 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 memfilter data secara dinamis 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 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 Antarmuka tanpa kode, ini adalah parameter kondisi filter.	Sintaks database Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari rentang waktu tertentu setiap hari.
MongoDB Reader	query Catatan Dalam Antarmuka tanpa kode, ini adalah parameter Search Condition.	Mirip dengan sintaks database Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari rentang waktu tertentu setiap hari.
OSS Reader	Object	Tentukan path objek Catatan Anda dapat menggunakan parameter ini bersama parameter penjadwalan untuk membaca data dari file tertentu setiap hari.
...	...	...

Konfigurasi 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 parameter placeholder dengan nilai aktual dari ekspresi parameter penjadwalan.

Sebagai contoh, untuk menyinkronkan data dari sumber data MySQL:

Jika Anda tidak mengonfigurasi Data Filtering, node tersebut secara default akan menyinkronkan data penuh ke tabel tujuan.
Jika Anda mengonfigurasi Data Filtering, node tersebut hanya akan menyinkronkan data yang memenuhi kondisi ke tabel tujuan.

Nama partisi tabel MaxCompute tujuan ditentukan dengan menggunakan parameter penjadwalan. Parameter $bizdate merepresentasikan Waktu data. Saat tugas terjadwal dijalankan, ekspresi partisi dalam konfigurasi tugas diganti dengan Waktu data. Untuk detail cara mengonfigurasi ekspresi parameter penjadwalan, lihat Konfigurasi dan penggunaan parameter penjadwalan. Untuk node sinkronisasi batch, Anda harus mengonfigurasi parameter bizdate di tiga lokasi untuk menerapkan sinkronisasi inkremental: Di bidang Data Filtering untuk 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 hari tersebut berdasarkan Waktu data. Di bidang Partition Information untuk tujuan, masukkan pt=${bizdate} untuk menulis data ke partisi tanggal yang sesuai, dan atur Cleanup Rule menjadi Insert Overwrite. Di bidang Parameters dalam panel Scheduling di sebelah kanan, masukkan bizdate=$bizdate. Hal ini memungkinkan sistem penjadwalan secara otomatis mengganti ${bizdate} dengan Waktu data aktual saat waktu proses. Saat Anda mengonfigurasi sinkronisasi data inkremental:

Untuk bidang inkremental bertipe data waktu atau tanggal, Anda dapat menggunakan parameter penjadwalan untuk mengganti nilai secara dinamis. Saat waktu proses, sistem secara otomatis mengganti parameter dengan nilai spesifik berdasarkan Waktu data. Untuk informasi lebih lanjut, lihat Format parameter penjadwalan.
Untuk bidang inkremental non-waktu: Anda dapat menggunakan assignment node untuk mengonversi bidang tersebut ke tipe data target, lalu meneruskannya ke Data Integration untuk sinkronisasi data. Untuk informasi lebih lanjut, lihat Assignment node.

Catatan

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

Keamanan Insert Overwrite: Menggunakan mode Insert Overwrite aman. Mode ini hanya membersihkan partisi yang ditentukan oleh tugas saat ini. Mode ini tidak memengaruhi data di partisi lain dan tidak menyebabkan konflik data atau penghapusan tidak disengaja.
Batasan overwrite massal untuk rentang partisi: DataWorks tidak mendukung pengaturan nilai rentang jam, seperti hh=00-23, dalam konfigurasi partisi untuk operasi overwrite massal. Untuk meng-overwrite data selama beberapa jam, Anda harus mengonfigurasi dan menjalankan tugas terpisah, karena parameter partisi hanya mendukung nilai tunggal atau karakter wildcard *.
Format sintaks wildcard: Dalam skenario di mana sumber berisi partisi per jam tetapi tujuan dipartisi hanya berdasarkan hari, Anda harus memasukkan karakter wildcard * di bidang partisi jam untuk mencocokkan semua data per jam. Masukkan * secara langsung. Jangan tambahkan tanda kutip, seperti "*", karena akan menyebabkan kesalahan sintaks.

Sinkronisasi inkremental berbasis timestamp dengan frekuensi tinggi

DataWorks mendukung sinkronisasi inkremental berbasis timestamp melalui node sinkronisasi batch yang dikombinasikan dengan penjadwalan periodik, misalnya setiap 5 menit atau setiap jam. Metode ini cocok untuk skenario sinkronisasi T+1 atau near-real-time dari sumber seperti ApsaraDB RDS for MySQL ke tujuan seperti SelectDB atau StarRocks. Pendekatan ini menggunakan pemfilteran SQL untuk muatan inkremental dan menghindari biaya menjalankan tugas CDC real-time terus-menerus. Poin konfigurasi utama:

Dalam klausa where bagian Data Filtering sumber, gunakan bidang 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. Hal ini memastikan setiap eksekusi terjadwal hanya menyinkronkan data inkremental dalam interval yang ditentukan. Untuk informasi lebih lanjut tentang konfigurasi parameter penjadwalan, lihat Kasus penggunaan parameter penjadwalan dalam Data Integration.
Dalam pemetaan bidang tujuan, tambahkan secara manual parameter konstan yang dipetakan ke bidang partisi untuk mengaktifkan penulisan partisi dinamis.

Konfigurasi inkremental untuk sinkronisasi batch seluruh database

Selain tugas sinkronisasi tabel tunggal, Anda dapat membuat tugas sinkronisasi batch seluruh database untuk mencapai sinkronisasi inkremental periodik. Saat membuat tugas, pilih sinkronisasi inkremental dan konfigurasikan kondisi inkremental dalam pengaturan tugas. Hal ini memungkinkan sinkronisasi partisi inkremental harian yang efisien, misalnya dengan memfilter berdasarkan bidang create_time.

Metode ini cocok untuk skenario di mana Anda ingin mengelola sinkronisasi beberapa tabel secara terpusat tetapi hanya melakukan pembaruan inkremental pada tabel tertentu. Untuk proses konfigurasi lengkap tugas sinkronisasi batch seluruh database, lihat Konfigurasi tugas sinkronisasi batch seluruh database.

Kasus penggunaan

Menyinkronkan data historis: Untuk menyinkronkan data inkremental historis ke partisi berbasis waktu yang sesuai dalam tabel tujuan, Anda dapat menggunakan fitur pengisian ulang data di Operation Center. Untuk detail cara menggunakan fitur ini, lihat Mengelola instans pengisian ulang data. Dalam konfigurasi node, pilih MySQL sebagai sumber data dan MaxCompute sebagai tujuan, dengan nama tabel seperti czd. Dalam kondisi Data Filtering, gunakan ${bizdate} untuk mengontrol rentang inkremental, misalnya, STR_TO_DATE('${bizdate}','%Y%m%d') <= gmt_modify_time. Atur Partition Information menjadi ds=${bizdate} dan pilih Insert Overwrite sebagai aturan pembersihan. Di bagian Parameters dalam konfigurasi penjadwalan, definisikan bizdate=$bizdate. Dalam skenario pengisian ulang data, sistem penjadwalan secara otomatis mengganti parameter ini dengan tanggal yang sesuai dari Waktu data. Saat Anda menjalankan pengisian ulang data, Anda dapat mengatur beberapa rentang Waktu data, seperti dari 2022-05-01 hingga 2022-05-31 dan dari 2022-04-01 hingga 2022-04-30. Pilih Run Backfill Instances with Schedule Time Later Than Current Time Immediately dan pilih Ascending Order of Data Timestamp untuk urutan eksekusi.
Menyinkronkan data secara inkremental dari ApsaraDB RDS ke MaxCompute

FAQ

MaxCompute Reader: kesalahan "partition does not exist"

Penyebab: parameter penjadwalan yang dikonfigurasi tidak diurai dengan benar menjadi nilai partisi aktual saat waktu proses, atau nilai yang diurai tidak cocok dengan partisi aktual di tabel sumber.

Solusi: Jika nilai partisi diteruskan dari parameter outputs node hulu, periksa konfigurasi parameter di DataStudio dan pastikan hal berikut:

Pastikan nama parameter yang dikonfigurasi sesuai dengan nama parameter input.
Pastikan nilai parameter yang diteruskan persis sesuai dengan partisi yang ada di MaxCompute.

Perilaku sinkronisasi default dan tabel non-partisi

Perilaku default

Secara default, tugas sinkronisasi batch Data Integration DataWorks melakukan sinkronisasi penuh. Untuk mencapai sinkronisasi inkremental, Anda harus mengonfigurasi kondisi Data Filtering dan menggabungkannya dengan parameter penjadwalan.

Penanganan tabel sumber tanpa bidang partisi

Jika tabel database sumber, seperti tabel di ApsaraDB RDS, tidak memiliki bidang waktu atau partisi, Anda tidak dapat langsung memfilter data inkremental menggunakan klausa WHERE. Kami menyarankan agar Anda menambahkan bidang waktu, seperti dt atau gmt_modify_time, ke tabel sumber untuk digunakan sebagai dasar pemfilteran inkremental. Setelah menyiapkan bidang tersebut, rujuk bagian 'Konfigurasi sinkronisasi inkremental' dalam dokumen ini untuk mengonfigurasi logika sinkronisasi inkremental.

splitPk dan splitFactor: Tujuan dan dampak

Definisi parameter

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

Dampak performa dan rekomendasi

Mengaktifkan splitPk dan splitFactor meningkatkan beban pada database sumber. Untuk mengurangi beban ini, kami menyarankan agar Anda:

Mengurangi konkurensi, atau mengatur konkurensi untuk tugas batch seluruh database menjadi 1.
Memastikan bahwa kolom pemisahan (splitPk) diindeks untuk meningkatkan efisiensi baca.