All Products
Search

This Product

    ApsaraDB for OceanBase (Deprecated):Menyinkronkan data dari database OceanBase ke instans Kafka

    Document Center

    ApsaraDB for OceanBase (Deprecated):Menyinkronkan data dari database OceanBase ke instans Kafka

    Last Updated:Dec 30, 2025

    Topik ini menjelaskan cara menggunakan layanan transmisi data untuk menyinkronkan data dari database OceanBase ke instans Kafka.

    Latar Belakang

    Kafka adalah platform komputasi aliran terdistribusi berkinerja tinggi yang banyak digunakan. Layanan transmisi data mendukung sinkronisasi data real-time antara instans Kafka self-managed dan database OceanBase dalam mode kompatibel Oracle atau MySQL, sehingga memperluas kemampuan pemrosesan pesan. Oleh karena itu, layanan ini banyak diterapkan dalam skenario bisnis seperti pembuatan gudang data real-time, kueri data, dan distribusi laporan.

    Prasyarat

    • Layanan transmisi data memiliki hak istimewa untuk mengakses sumber daya cloud. Untuk informasi selengkapnya, lihat Berikan hak istimewa kepada peran untuk transmisi data.

    • Anda telah membuat pengguna database khusus untuk sinkronisasi data di database OceanBase sumber dan memberikan hak istimewa yang sesuai kepada pengguna tersebut. Untuk informasi selengkapnya, lihat Buat pengguna database.

    Batasan

    • Hanya tabel fisik yang dapat disinkronkan.

    • Layanan transmisi data mendukung Kafka 0.9, 1.0, dan 2.x.

    • Selama sinkronisasi data, jika Anda mengganti nama tabel sumber yang akan disinkronkan dan nama barunya berada di luar cakupan sinkronisasi, data dari tabel sumber tersebut tidak akan disinkronkan ke instans Kafka target.

    • Nama tabel yang akan disinkronkan, serta nama kolom dalam tabel tersebut, tidak boleh mengandung karakter Tionghoa.

    • Layanan transmisi data hanya mendukung migrasi objek jika kondisi berikut terpenuhi: nama database, nama tabel, dan nama kolom objek tersebut dikodekan dalam ASCII tanpa karakter khusus. Karakter khusus tersebut meliputi line break, spasi, dan karakter berikut: . | " ' ` ( ) = ; / & \.

    • Layanan transmisi data tidak mendukung database OceanBase standby sebagai sumber.

    Pertimbangan

    • Pada tugas sinkronisasi data dengan sumber berupa database OceanBase dan sinkronisasi DDL diaktifkan, jika operasi RENAME dilakukan pada tabel di database sumber, kami menyarankan Anda untuk me-restart task guna menghindari kehilangan data selama sinkronisasi inkremental.

    • Jika Anda menggunakan OceanBase Database V4.x dan telah memilih sinkronisasi inkremental, Anda perlu menentukan atribut STORED untuk kolom yang dihasilkan. Jika Anda tidak menentukan atribut ini, informasi tentang kolom yang dihasilkan tidak akan disimpan dalam log inkremental, yang dapat menyebabkan kesalahan data pada sinkronisasi inkremental.

    • Perhatikan pertimbangan berikut ketika baris yang diperbarui berisi kolom large object (LOB):

      • Jika kolom LOB diperbarui, jangan gunakan nilai yang disimpan di kolom LOB sebelum operasi UPDATE atau DELETE.

        Tipe data berikut disimpan dalam kolom LOB: JSON, GIS, XML, user-defined type (UDT), dan TEXT seperti LONGTEXT dan MEDIUMTEXT.

      • Jika kolom LOB tidak diperbarui, nilai yang disimpan di kolom LOB sebelum dan sesudah operasi UPDATE atau DELETE adalah NULL.

    • Jika jam antar node atau antara client dan server tidak tersinkronisasi, latensi selama sinkronisasi inkremental mungkin tidak akurat.

      Misalnya, jika jam lebih awal dari waktu standar, latensi bisa bernilai negatif. Jika jam lebih lambat dari waktu standar, latensi bisa bernilai positif.

    • Ketika transfer data dilanjutkan untuk suatu tugas, beberapa data (dalam rentang satu menit terakhir) mungkin menjadi duplikat di instans Kafka. Oleh karena itu, deduplikasi diperlukan di sistem downstream.

    • Ketika Anda menyinkronkan data dari database OceanBase ke instans Kafka, jika pernyataan pembuatan indeks unik gagal dieksekusi di sumber, instans Kafka tetap mengonsumsi pernyataan DDL untuk pembuatan dan penghapusan indeks unik tersebut. Jika pernyataan DDL downstream untuk pembuatan indeks unik gagal dieksekusi, abaikan pengecualian ini.

    • Jika Anda hanya memilih Incremental Synchronization saat membuat tugas sinkronisasi data, layanan transmisi data mensyaratkan agar log inkremental lokal di database sumber dipertahankan setidaknya selama 48 jam.

      Jika Anda memilih Full Synchronization dan Incremental Synchronization saat membuat tugas sinkronisasi data, layanan transmisi data mensyaratkan agar log inkremental lokal di database sumber dipertahankan setidaknya selama tujuh hari. Jika tidak, tugas sinkronisasi data mungkin gagal atau data di sumber dan target menjadi tidak konsisten karena layanan transmisi data tidak dapat memperoleh log inkremental.

    Tipe instans sumber dan target yang didukung

    Pada tabel berikut, OB_MySQL merepresentasikan database OceanBase dalam mode kompatibel MySQL, dan OB_Oracle merepresentasikan database OceanBase dalam mode kompatibel Oracle.

    Sumber

    Target

    OB_MySQL (instans kluster OceanBase)

    Kafka (instans Kafka di Alibaba Cloud)

    OB_MySQL (instans kluster OceanBase)

    Kafka (instans Kafka self-managed di VPC)

    OB_MySQL (instans kluster OceanBase)

    Kafka (instans Kafka di jaringan publik)

    OB_MySQL (instans serverless)

    Kafka (instans Kafka di Alibaba Cloud)

    OB_MySQL (instans serverless)

    Kafka (instans Kafka self-managed di VPC)

    OB_MySQL (instans serverless)

    Kafka (instans Kafka di jaringan publik)

    OB_Oracle (instans kluster OceanBase)

    Kafka (instans Kafka di Alibaba Cloud)

    OB_Oracle (instans kluster OceanBase)

    Kafka (instans Kafka self-managed di VPC)

    OB_Oracle (instans kluster OceanBase)

    Kafka (instans Kafka di jaringan publik)

    DDL yang Didukung

    • CREATE TABLE

      Penting

      Tabel yang dibuat harus merupakan objek sinkronisasi. Untuk mengeksekusi pernyataan CREATE TABLE pada tabel yang disinkronkan, eksekusi pernyataan DROP TABLE pada tabel tersebut terlebih dahulu.

    • ALTER TABLE

    • DROP TABLE

    • TRUNCATE TABLE

      Catatan

      Pada penghapusan tertunda, transaksi yang sama berisi dua pernyataan DDL TRUNCATE TABLE yang identik. Dalam kasus ini, idempotensi diimplementasikan untuk konsumsi downstream.

    • ALTER TABLE…TRUNCATE PARTITION

    • CREATE INDEX

    • DROP INDEX

    • COMMENT ON TABLE

    • RENAME TABLE

      Penting

      Tabel yang diganti namanya harus merupakan objek sinkronisasi.

    Prosedur

    1. Login ke

      ApsaraDB for OceanBase console

      Untuk informasi selengkapnya, lihat Beli tugas sinkronisasi data.

    2. Pilih Data Transmission > Data Synchronization. Pada halaman yang muncul, klik Configuration untuk tugas sinkronisasi data tersebut.

      image.png

      Jika Anda ingin merujuk konfigurasi tugas yang sudah ada, klik Reference Configuration. Untuk informasi selengkapnya, lihat Referensi dan hapus konfigurasi tugas sinkronisasi data.

    3. Pada halaman Select Source and Target, konfigurasikan parameter-parameter berikut.

      Parameter

      Deskripsi

      Synchronization Task Name

      Kami menyarankan Anda menetapkannya sebagai kombinasi angka dan huruf. Nama tidak boleh mengandung spasi dan panjangnya tidak boleh melebihi 64 karakter.

      Source

      Jika Anda telah membuat sumber data OceanBase, pilih dari daftar drop-down. Jika belum, klik New Data Source pada daftar drop-down dan buat di dialog yang muncul di sebelah kanan. Untuk informasi selengkapnya mengenai parameter, lihat Buat sumber data OceanBase.

      Penting

      Sumber tidak boleh berupa database OceanBase dengan Instance Type berupa OceanBase Database Tenant Instance.

      Target

      Jika Anda telah membuat sumber data Kafka, pilih dari daftar drop-down. Jika belum, klik New Data Source pada daftar drop-down dan buat di dialog yang muncul di sebelah kanan. Untuk informasi selengkapnya, lihat Buat sumber data Kafka.

      Tag (Opsional)

      Pilih tag target dari daftar drop-down. Anda juga dapat mengklik Manage Tags untuk membuat, mengubah, dan menghapus tag. Untuk informasi selengkapnya, lihat Gunakan tag untuk mengelola tugas sinkronisasi data.

    4. Klik Next. Pada halaman Select Synchronization Type, tentukan tipe sinkronisasi untuk tugas sinkronisasi data saat ini.

      image

      Tipe sinkronisasi yang didukung adalah Full Synchronization dan Incremental Synchronization. Incremental Synchronization mendukung DML Synchronization dan DDL Synchronization. Operasi DML yang didukung adalah INSERT, DELETE, dan UPDATE. Anda dapat memilih opsi sesuai kebutuhan. Untuk informasi selengkapnya, lihat Konfigurasikan sinkronisasi DDL/DML.

    5. Klik Next. Pada halaman Select Synchronization Objects, pilih objek yang akan disinkronkan dalam tugas sinkronisasi data saat ini.

      Anda dapat menggunakan opsi Specify Objects atau Match Rules untuk menentukan objek sinkronisasi. Topik ini menjelaskan cara menggunakan opsi Specify Objects untuk menentukan objek sinkronisasi. Untuk informasi tentang cara mengonfigurasi aturan pencocokan, lihat "Pola wildcard untuk migrasi/sinkronisasi data antara database dan instans Message Queue" dalam Konfigurasikan dan modifikasi aturan pencocokan.

      Catatan

      Jika Anda memilih DDL Synchronization pada langkah Select Synchronization Type, kami menyarankan Anda memilih objek sinkronisasi menggunakan opsi Match Rules. Hal ini memastikan bahwa semua objek baru yang memenuhi aturan pencocokan akan disinkronkan. Jika Anda memilih objek sinkronisasi menggunakan opsi Specify Objects, objek baru atau objek yang diganti namanya tidak akan disinkronkan.

      Saat Anda menyinkronkan data dari database OceanBase ke instans Kafka, Anda dapat menyinkronkan data dari beberapa tabel ke beberapa topik.

      1. Pada bagian Select Synchronization Objects, pilih Specify Objects.

      2. Di panel sebelah kiri, pilih objek yang akan disinkronkan.

      3. Klik >.

      4. Klik daftar drop-down Existing Topics di dialog Map Object to Topic dan pilih topik target.

        image.png

      5. Klik OK.

        Layanan transmisi data memungkinkan Anda mengimpor objek menggunakan teks. Layanan ini juga memungkinkan Anda mengubah topik objek, menetapkan filter baris, dan menghapus satu objek atau semua objek. Objek di database target ditampilkan dalam struktur Topik > Database > Tabel.

        Catatan

        Saat Anda memilih Match Rules untuk menentukan objek sinkronisasi, penggantian nama objek diimplementasikan berdasarkan sintaks aturan pencocokan yang ditentukan. Di area operasi, Anda hanya dapat menetapkan kondisi filter dan memilih kolom sharding serta kolom yang akan disinkronkan. Untuk informasi selengkapnya, lihat Konfigurasikan dan modifikasi aturan pencocokan.

        image.png

        Operasi

        Deskripsi

        Import objects

        1. Di daftar sebelah kanan, klik Import Objects di pojok kanan atas.

        2. Di dialog yang muncul, klik OK.

          Penting

          Operasi ini akan menimpa pilihan sebelumnya. Lakukan dengan hati-hati.

        3. Di dialog Import Synchronization Objects, impor objek yang akan disinkronkan. Anda dapat mengimpor file CSV untuk menetapkan kondisi filter baris, kolom filter, dan kolom sharding. Untuk informasi selengkapnya, lihat Unduh dan impor pengaturan objek sinkronisasi.

        4. Klik Validate.

        5. Setelah validasi berhasil, klik OK.

        Change topics

        Layanan transmisi data memungkinkan Anda mengubah topik untuk objek di target. Untuk informasi selengkapnya, lihat Ubah topik.

        Configure settings

        Anda dapat menggunakan klausa WHERE untuk memfilter data berdasarkan baris dan memilih kolom sharding serta kolom yang akan disinkronkan.

        Di dialog Settings, Anda dapat melakukan operasi berikut:

        • Di bagian Row Filters, tentukan klausa WHERE SQL standar untuk memfilter data berdasarkan baris. Untuk informasi selengkapnya, lihat Gunakan kondisi SQL untuk memfilter data.

        • Pilih kolom sharding yang ingin Anda gunakan dari daftar drop-down Sharding Columns. Anda dapat memilih beberapa bidang sebagai kolom sharding. Parameter ini bersifat opsional.

          Kecuali ditentukan lain, pilih primary key sebagai kolom sharding. Jika primary key tidak seimbang beban, pilih bidang yang seimbang beban dengan pengenal unik sebagai kolom sharding untuk menghindari potensi masalah kinerja. Kolom sharding dapat digunakan untuk tujuan berikut:

          • Load balancing: Thread yang digunakan untuk mengirim pesan dapat dikenali berdasarkan kolom sharding jika tabel target mendukung penulisan konkuren.

          • Orderliness: Layanan transmisi data memastikan bahwa pesan diterima secara berurutan jika nilai kolom sharding sama. Orderliness menentukan urutan eksekusi pernyataan DML untuk suatu kolom.

        • Di bagian Select Columns, pilih kolom yang akan disinkronkan. Untuk informasi selengkapnya, lihat Filtering kolom.

        Remove one or all objects

        Layanan transmisi data memungkinkan Anda menghapus satu atau semua objek sinkronisasi yang ditambahkan ke daftar sebelah kanan selama pemetaan data.

        • Hapus satu objek sinkronisasi

          Di daftar sebelah kanan, arahkan kursor ke objek yang ingin dihapus, lalu klik Remove untuk menghapus objek sinkronisasi tersebut.

        • Hapus semua objek sinkronisasi

          Di daftar sebelah kanan, klik Remove All di pojok kanan atas. Di dialog yang muncul, klik OK untuk menghapus semua objek sinkronisasi.

    6. Klik Next. Pada halaman Synchronization Options, konfigurasikan parameter-parameter berikut.

      • Full synchronization

        Tabel berikut menjelaskan parameter sinkronisasi penuh, yang hanya ditampilkan jika Anda telah memilih Full Synchronization pada halaman Select Synchronization Type.

        image

        Parameter

        Deskripsi

        Read Concurrency

        Konkurensi untuk membaca data dari sumber selama sinkronisasi penuh. Nilai maksimum adalah 512. Konkurensi tinggi dapat menyebabkan beban berlebih pada sumber, sehingga memengaruhi bisnis.

        Write Concurrency

        Konkurensi untuk menulis data ke target selama sinkronisasi penuh. Nilai maksimum adalah 512. Konkurensi tulis tinggi dapat menyebabkan beban berlebih pada target, sehingga memengaruhi bisnis.

        Full Synchronization Rate Limit

        Anda dapat memilih apakah akan membatasi laju sinkronisasi penuh sesuai kebutuhan. Jika Anda memilih untuk membatasi laju sinkronisasi penuh, Anda harus menentukan records per second (RPS) dan bytes per second (BPS). RPS menentukan jumlah maksimum baris data yang disinkronkan ke target per detik selama sinkronisasi penuh, sedangkan BPS menentukan jumlah maksimum data dalam byte yang disinkronkan ke target per detik selama sinkronisasi penuh.

        Catatan

        Nilai RPS dan BPS yang ditentukan di sini hanya untuk Pembatasan kecepatan. Kinerja sinkronisasi penuh aktual bergantung pada faktor-faktor seperti pengaturan sumber dan target serta spesifikasi instans.

      • Incremental synchronization

        Tabel berikut menjelaskan parameter sinkronisasi inkremental, yang hanya ditampilkan jika Anda telah memilih Incremental Synchronization pada halaman Select Synchronization Type.

        image

        Parameter

        Deskripsi

        Write Concurrency

        Konkurensi untuk menulis data ke target selama sinkronisasi inkremental. Nilai maksimum adalah 512. Konkurensi tulis tinggi dapat menyebabkan beban berlebih pada target, sehingga memengaruhi bisnis.

        Incremental Synchronization Rate Limit

        Anda dapat memilih apakah akan membatasi laju sinkronisasi inkremental sesuai kebutuhan. Jika Anda memilih untuk membatasi laju sinkronisasi inkremental, Anda harus menentukan RPS dan BPS. RPS menentukan jumlah maksimum baris data yang disinkronkan ke target per detik selama sinkronisasi inkremental, sedangkan BPS menentukan jumlah maksimum data dalam byte yang disinkronkan ke target per detik selama sinkronisasi inkremental.

        Catatan

        Nilai RPS dan BPS yang ditentukan di sini hanya untuk Pembatasan kecepatan. Kinerja sinkronisasi inkremental aktual bergantung pada faktor-faktor seperti pengaturan sumber dan target serta spesifikasi instans.

        Incremental Synchronization Start Timestamp

        • Parameter ini tidak tersedia jika Anda telah memilih Full Synchronization.

        • Jika Anda telah memilih Incremental Synchronization tetapi tidak memilih Full Synchronization, tentukan titik waktu setelah data akan disinkronkan. Nilai default adalah waktu sistem saat ini. Untuk informasi selengkapnya, lihat Tetapkan timestamp sinkronisasi inkremental.

      • Advanced parameters

        image

        Parameter

        Deskripsi

        Serialization Method

        Format pesan untuk menyinkronkan data ke instans Kafka target. Nilai yang valid adalah Default, Canal, DataWorks (versi 2.0 didukung), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten, DebeziumSmt, dan Avro. Untuk informasi selengkapnya, lihat Format data.

        Penting

        • Hanya database OceanBase dalam mode kompatibel MySQL yang mendukung Debezium, DebeziumFlatten, DebeziumSmt, dan Avro.

        • Jika format pesan diatur ke DataWorks, operasi DDL COMMENT ON TABLE dan ALTER TABLE…TRUNCATE PARTITION tidak dapat disinkronkan.

        Partitioning Rules

        Aturan untuk menyinkronkan data dari database OceanBase ke topik Kafka. Layanan transmisi data mendukung Hash, Table, dan One. Untuk informasi selengkapnya mengenai pengiriman pernyataan DDL dalam berbagai skenario dan contoh, lihat deskripsi di bawah.

        • Hash menunjukkan bahwa layanan transmisi data menggunakan algoritma hash untuk memilih partisi topik Kafka berdasarkan nilai primary key atau kolom sharding.

        • Table menunjukkan bahwa layanan transmisi data mengirimkan semua data dalam tabel ke partisi yang sama dan menggunakan nama tabel sebagai kunci hash.

        • One menunjukkan bahwa pesan JSON dikirimkan ke satu partisi dalam topik untuk memastikan pengurutan.

        Business System Identification (Opsional)

        Mengidentifikasi sistem bisnis sumber data. Parameter ini hanya ditampilkan jika Anda memilih DataWorks untuk Serialization Method. Pengenal sistem bisnis terdiri dari 1 hingga 20 karakter.

        Tabel berikut menjelaskan pengiriman pernyataan DDL dalam berbagai skenario.

        Aturan partisi

        Pernyataan DDL yang melibatkan beberapa tabel

        (contoh: RENAME TABLE)

        Pernyataan DDL yang melibatkan tabel tidak dikenal

        (contoh: DROP INDEX)

        Pernyataan DDL yang melibatkan satu tabel

        Hash

        Pernyataan DDL dikirimkan ke semua partisi topik yang terkait dengan tabel yang terlibat.

        Asumsikan pernyataan DDL melibatkan tiga tabel: A, B, dan C. Jika A terkait dengan Topik 1, B terkait dengan Topik 2, dan C tidak terlibat dalam tugas saat ini, pernyataan DDL dikirimkan ke semua partisi Topik 1 dan Topik 2.

        Pernyataan DDL dikirimkan ke semua partisi semua topik dalam tugas saat ini.

        Asumsikan pernyataan DDL tidak dapat diidentifikasi oleh layanan transmisi data. Jika tugas saat ini memiliki tiga topik, pernyataan DDL dikirimkan ke semua partisi ketiga topik tersebut.

        Pernyataan DDL dikirimkan ke semua partisi topik yang terkait dengan tabel tersebut.

        Table

        Pernyataan DDL dikirimkan ke partisi tertentu dari topik yang terkait dengan tabel. Partisi tersebut sesuai dengan nilai hash dari nama tabel yang terlibat.

        Asumsikan pernyataan DDL melibatkan tiga tabel: A, B, dan C. Jika A terkait dengan Topik 1, B terkait dengan Topik 2, dan C tidak terlibat dalam tugas saat ini, pernyataan DDL dikirimkan ke partisi yang sesuai dengan nilai hash nama tabel yang terlibat di Topik 1 dan Topik 2.

        Pernyataan DDL dikirimkan ke semua partisi semua topik dalam tugas saat ini.

        Asumsikan pernyataan DDL tidak dapat diidentifikasi oleh layanan transmisi data. Jika tugas saat ini memiliki tiga topik, pernyataan DDL dikirimkan ke semua partisi ketiga topik tersebut.

        Pernyataan DDL dikirimkan ke satu partisi topik yang terkait dengan tabel tersebut.

        One

        Pernyataan DDL dikirimkan ke partisi tetap dari topik yang terkait dengan tabel.

        Asumsikan pernyataan DDL melibatkan tiga tabel: A, B, dan C. Jika A terkait dengan Topik 1, B terkait dengan Topik 2, dan C tidak terlibat dalam tugas saat ini, pernyataan DDL dikirimkan ke partisi tetap Topik 1 dan Topik 2.

        Pernyataan DDL dikirimkan ke partisi tetap semua topik dalam tugas saat ini.

        Asumsikan pernyataan DDL tidak dapat diidentifikasi oleh layanan transmisi data. Jika tugas saat ini memiliki tiga topik, pernyataan DDL dikirimkan ke partisi tetap ketiga topik tersebut.

        Pernyataan DDL dikirimkan ke partisi tetap topik yang terkait dengan tabel tersebut.

    7. Klik Precheck.

      Selama precheck, layanan transmisi data mendeteksi koneksi antara sumber dan target. Jika terjadi error selama precheck, Anda dapat melakukan operasi berikut:

      • Identifikasi dan atasi masalah tersebut, lalu lakukan precheck lagi.

      • Klik Skip di kolom Actions pada item precheck yang gagal. Di dialog yang menampilkan konsekuensi operasi tersebut, klik OK.

    8. Setelah precheck berhasil, klik Start Task.

      Jika Anda tidak perlu memulai tugas sekarang, klik Save. Anda dapat memulai tugas secara manual di halaman Synchronization Tasks atau melalui operasi batch nanti. Untuk informasi selengkapnya mengenai operasi batch, lihat Lakukan operasi batch pada tugas sinkronisasi data.

      Layanan transmisi data memungkinkan Anda memodifikasi objek sinkronisasi saat tugas sinkronisasi sedang berjalan. Untuk informasi selengkapnya, lihat Lihat dan modifikasi objek sinkronisasi serta kondisi filternya. Setelah tugas sinkronisasi data dimulai, tugas tersebut akan dieksekusi berdasarkan tipe sinkronisasi yang dipilih. Untuk informasi selengkapnya, lihat Lihat detail tugas sinkronisasi data.

    Jika tugas sinkronisasi data mengalami pengecualian eksekusi akibat kegagalan jaringan atau startup proses yang lambat, Anda dapat mengklik Resume di halaman Synchronization Tasks atau Details tugas sinkronisasi tersebut.

    Referensi