Realtime Compute for Apache Flink:OceanBase

Parameter umum

Parameter berikut berlaku untuk semua jenis tabel.

Parameter tabel sumber

Parameter tabel dimensi

Parameter tabel sink (JDBC)

Parameter tabel sink (impor bypass)

Impor bypass adalah metode penulisan throughput tinggi untuk pemuatan data bulk di OceanBase. Tersedia di VVR 11.5 dan versi lebih baru.

Sebelum menggunakan impor bypass, periksa batasan berikut:

• Hanya aliran bounded: Sumber data harus berupa aliran bounded. Gunakan mode batch Flink untuk performa terbaik.

• Penguncian tabel selama impor: Tabel tujuan dikunci selama proses impor. Penulisan DML dan perubahan DDL diblokir; kueri baca tidak terpengaruh.

• Bukan untuk penulisan real-time: Untuk penulisan streaming atau real-time, gunakan sink JDBC.

Mode kompatibel MySQL

Mode kompatibel Oracle

Tabel sumber dan tabel sink

Contoh berikut membaca data CDC dari tabel sumber OceanBase dan menulisnya ke tabel sink JDBC. Definisi tabel sink impor bypass juga disertakan sebagai referensi.

Ketiga tabel menggunakan 'connector' = 'oceanbase'. Tabel sumber dan sink JDBC menggunakan set parameter yang berbeda; sink direct-load mengatur sink.mode = 'direct-load' dan terhubung melalui port RPC.

-- Tabel sumber CDC OceanBase (membaca riwayat lengkap, lalu Binlog)
CREATE TEMPORARY TABLE oceanbase_source (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'     = 'oceanbase',
  'hostname'      = '<your-hostname>',
  'port'          = '3306',
  'username'      = '<your-username>',
  'password'      = '<your-password>',
  'database-name' = '<your-database-name>',
  'table-name'    = '<your-table-name>'
);

-- Tabel sink JDBC OceanBase (untuk penulisan streaming real-time)
CREATE TEMPORARY TABLE oceanbase_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '<your-password>',
  'tableName' = '<your-table-name>'
);

-- Tabel sink impor bypass OceanBase (untuk penulisan batch throughput tinggi)
-- Memerlukan sumber data bounded; atur Flink ke mode batch untuk performa terbaik
CREATE TEMPORARY TABLE oceanbase_directload_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'   = 'oceanbase',
  'sink.mode'   = 'direct-load',
  'host'        = '<your-host>',
  'port'        = '<your-rpc-port>',
  'tenant-name' = '<your-tenant-name>',
  'schema-name' = '<your-schema-name>',
  'table-name'  = '<your-table-name>',
  'username'    = '<your-username>',
  'password'    = '<your-password>'
);

BEGIN STATEMENT SET;
INSERT INTO oceanbase_sink
SELECT * FROM oceanbase_source;
END;

Tabel dimensi

Contoh berikut melakukan join antara sumber Datagen dengan tabel dimensi OceanBase menggunakan temporal join. Kebijakan cache ALL memuat seluruh tabel dimensi ke memori sebelum pekerjaan dimulai.

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

-- Tabel dimensi OceanBase dengan kebijakan cache ALL
-- Cache ALL memuat tabel penuh saat startup — cocok untuk tabel kecil dan stabil
CREATE TEMPORARY TABLE oceanbase_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '${secret_values.password}',
  'tableName' = '<your-table-name>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN oceanbase_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H
ON T.a = H.a;