DataWorks:Sinkronisasi real-time dari tabel Hologres ke Kafka

1. Pilih jenis tugas sinkronisasi

1. Masuk ke Konsol DataWorks. Di wilayah target, klik Data Integration > Data Integration di panel navigasi sebelah kiri. Pilih ruang kerja dari daftar drop-down dan klik Go to Data Integration.

2. Di panel navigasi sebelah kiri, klik Sync Tasks. Di bagian atas halaman, klik Create synchronization task dan konfigurasikan parameter berikut.

   • Data Source And Destination: Hologres→Kafka

   • New Task Name: Sesuaikan nama untuk tugas sinkronisasi.

   • Synchronization Mode: single-table real-time.

   • Synchronization Mode Step: Pilih full synchronization.

2. Konfigurasi jaringan dan sumber daya

1. Di bagian Configure Network Connections and Resource Groups, pilih Resource Group untuk tugas sinkronisasi. Anda juga dapat mengatur Task Resource Usage dalam CUs.

2. Untuk Source Data Source, pilih sumber data Hologres. Untuk Destination data source, pilih sumber data Kafka. Lalu, klik Test Connection.

3. Setelah pengujian konektivitas untuk sumber dan tujuan berhasil, klik Next step.

3. Konfigurasi tautan sinkronisasi

4. Peringatan

Untuk mencegah kesalahan tugas menyebabkan keterlambatan dalam sinkronisasi data bisnis, Anda dapat menetapkan kebijakan peringatan untuk tugas sinkronisasi.

1. Klik Alert Settings di pojok kanan atas halaman untuk membuka halaman pengaturan Alert Rule Configurations for Real-time Synchronization Subnode.

2. Klik Add Alert Rule untuk mengonfigurasi aturan peringatan.

   Catatan

   Aturan peringatan yang Anda definisikan di sini berlaku untuk subtugas sinkronisasi real-time yang dihasilkan oleh tugas ini. Setelah Anda mengonfigurasi tugas, Anda dapat melihat dan mengubah aturan peringatan untuk subtugas tersebut di halaman Jalankan dan kelola tugas sinkronisasi real-time.

3. Kelola aturan peringatan.

   Untuk aturan peringatan yang sudah ada, Anda dapat menggunakan sakel toggle untuk mengaktifkan atau menonaktifkannya. Anda juga dapat mengirim peringatan ke penerima berbeda berdasarkan tingkat peringatan.

5. Pengaturan lanjutan

Tugas sinkronisasi menyediakan beberapa parameter yang dapat Anda ubah sesuai kebutuhan.

1. Klik advanced settings di pojok kanan atas halaman untuk membuka halaman pengaturan lanjutan.

2. Di halaman advanced settings, ubah nilai parameter sesuai kebutuhan.

6. Kelompok sumber daya

Anda dapat mengklik Configure Resource Group di pojok kanan atas untuk melihat dan mengganti kelompok sumber daya saat ini untuk tugas.

7. Jalankan tugas sinkronisasi

1. Setelah menyelesaikan semua pengaturan, klik Complete di bagian bawah halaman.

2. Di halaman Data Integration > Synchronization Task, temukan tugas yang Anda buat dan klik Start di kolom Operations.

3. Klik Name/ID tugas yang sesuai di Task List untuk melihat proses eksekusi terperincinya.

Lihat status tugas sinkronisasi

Setelah tugas sinkronisasi dibuat, buka halaman Tasks untuk melihat semua tugas sinkronisasi di ruang kerja dan informasi dasarnya.

Daftar tugas menampilkan kolom berikut: Name/ID, Data Source Synchronization Solution (jenis sinkronisasi, sumber, dan tujuan), Status, Execution Overview, resource group, dan Actions.

• Anda dapat Start atau Stop tugas sinkronisasi di kolom Actions. Di bawah More, Anda dapat melakukan operasi lain, seperti Edit dan View.

• Untuk tugas yang sedang berjalan, Anda dapat melihat status dasar di Execution Overview dan klik area ikhtisar yang sesuai untuk melihat detail eksekusi.

Tugas sinkronisasi real-time dari tabel Hologres ke Kafka terdiri dari tiga langkah berikut:

• Structure Migration: mencakup metode pembuatan tabel tujuan (tabel yang sudah ada atau pembuatan tabel otomatis). Jika Anda memilih pembuatan tabel otomatis, pernyataan bahasa definisi data (DDL) untuk membuat tabel akan ditampilkan.

• Full initialization: Jika Anda memilih Full Sync untuk langkah sinkronisasi tugas Anda, kemajuan inisialisasi penuh akan ditampilkan di sini.

• Real-time Data Synchronization: Berisi statistik tentang sinkronisasi real-time, termasuk trafik baca/tulis real-time, data rusak, failover, dan log operasi.

Jalankan ulang tugas

Dalam kasus khusus, seperti ketika Anda perlu mengubah bidang yang disinkronkan atau menyesuaikan informasi tabel target, Anda dapat mengklik Rerun di kolom Operations tugas sinkronisasi. Tindakan ini menyinkronkan bidang yang telah disesuaikan dan perubahan lainnya ke target. Proses ini melewati tabel yang tidak berubah dan sebelumnya telah disinkronkan.

• Untuk menjalankan ulang tugas tanpa perubahan apa pun, klik Rerun.

• Jika Anda mengedit tugas, klik Complete setelah melakukan perubahan. Aksi tugas berubah menjadi Apply Updates. Mengklik Apply Updates akan menjalankan ulang tugas dengan konfigurasi baru.

Canal CDC

Canal CDC adalah format data CDC yang didefinisikan oleh Alibaba Canal.

• Bidang dan maknanya

  Nama bidang	Makna nilai bidang
  id	Nilai bidang ini tetap 0.
  database	Nama database Hologres.
  table	Nama tabel Hologres.
  pkNames	Kolom kunci primer tabel Hologres.
  isDdl	Menentukan apakah log biner mencatat perubahan DDL. Nilai bidang ini tetap false karena sinkronisasi pesan untuk perubahan DDL pada tabel Hologres tidak didukung.
  type	Jenis perubahan DML. Nilai yang valid: INSERT, UPDATE, dan DELETE. Catatan Satu operasi perubahan pada tabel Hologres menghasilkan dua catatan bertipe UPDATE. Catatan-catatan tersebut akan ditulis ke topik Kafka. Salah satu catatan sesuai dengan konten data sebelum perubahan. Yang lainnya sesuai dengan konten data setelah perubahan. Jika Anda ingin menyinkronkan data lengkap dari tabel Hologres ke topik Kafka, nilai bidang type tetap INSERT.
  es	Timestamp 13 digit dalam milidetik. Timestamp menunjukkan waktu ketika data di tabel Hologres berubah. Jika Anda ingin menyinkronkan data lengkap dari tabel Hologres ke topik Kafka, nilai bidang es tetap 0.
  ts	Timestamp 13 digit dalam milidetik. Timestamp menunjukkan waktu ketika log biner yang dihasilkan untuk tabel Hologres dibaca oleh tugas sinkronisasi.
  sql	Kode SQL yang mencatat perubahan DDL ketika log biner yang dihasilkan untuk tabel Hologres berisi perubahan DDL. Nilai bidang ini tetap string kosong karena sinkronisasi pesan untuk perubahan DDL pada tabel Hologres tidak didukung.
  sqlType	Tipe data bidang SQL yang sesuai dengan tipe data bidang tabel Hologres. Pemetaan antara tipe data Hologres dan nilai sqlType yang valid: bigint: -5 decimal dengan skala bukan 0: 3 decimal dengan skala 0: -5 boolean: 16 date: 91 float4: 6 float8: 8 integer: 4 smallint: 5 json: 12 text: 12 varchar: 12 timestamp: 93 timestamptz: 93 bigserial: -5 bytea: 12 char: 12 serial: 4 time: 92 int4[]: 12 int8[]: 12 float4[]: 12 float8[]: 12 boolean[]: 12 text[]: 12
  mysqlType	Tipe data bidang MySQL yang sesuai dengan tipe data bidang tabel Hologres. Pemetaan antara tipe data Hologres dan nilai mysqlType yang valid: bigint: BIGINT int4: INT decimal dengan skala bukan 0: DECIMAL(xx,xx) decimal dengan skala 0: BIGINT boolean: BOOLEAN date: DATE float4: FLOAT float8: DOUBLE integer: INT smallint: SMALLINT json: TEXT text: TEXT varchar: VARCHAR(xx) timestamp: DATETIME(6) timestamptz: DATETIME(6) bigserial: BIGINT bytea: TEXT char: TEXT serial: INT time: TIME(6) int4[]: TEXT int8[]: TEXT float4[]: TEXT float8[]: TEXT boolean[]: TEXT text[]: TEXT
  data	Perubahan data di tabel Hologres. Nama bidang di tabel Hologres digunakan sebagai kunci, dan perubahan data di bidang tersebut diserialisasi menjadi string dan digunakan sebagai nilai. Kemudian, kunci dan nilai tersebut diatur sebagai string berformat JSON. Untuk informasi selengkapnya tentang serialisasi, lihat Deskripsi serialisasi JSON.
  old	Dua catatan bertipe UPDATE dihasilkan untuk satu operasi perubahan pada tabel Hologres dan ditulis ke topik Kafka. Catatan tersebut sesuai dengan konten data sebelum dan sesudah perubahan. Di catatan pertama, bidang old digunakan untuk mencatat konten data di tabel Hologres sebelum perubahan. Bidang data digunakan untuk mencatat konten data untuk perubahan DML selain UPDATE.

• Contoh data berformat JSON Canal yang sesuai dengan perubahan data yang dihasilkan oleh operasi INSERT di log biner tabel Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Contoh data berformat JSON Canal yang sesuai dengan data lengkap yang disinkronkan dari tabel Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 0,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Contoh dua catatan data berformat JSON Canal yang sesuai dengan perubahan data yang dihasilkan oleh operasi UPDATE di log biner tabel Hologres

  // Konten data sebelum perubahan
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "old": [
          {
              "bigint": "0",
              "integer": "0",
              "smallint": "0"
          }
      ],
      "data": null
  }
  // Konten data setelah perubahan
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Contoh data berformat JSON Canal yang sesuai dengan perubahan data yang dihasilkan oleh operasi DELETE di log biner tabel Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "DELETE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

Json

Format JSON menggunakan nama bidang log biner Hologres sebagai kunci dan menserialisasi nilai bidang menjadi string. Pasangan kunci-nilai yang dihasilkan ditulis ke topik Kafka sebagai string berformat JSON.