Realtime Compute for Apache Flink:OceanBase

Arka plan bilgileri

OceanBase, yerel dağıtık hibrit işlem (HTAP) veritabanı yönetim sistemidir. Daha fazla bilgi için OceanBase web sitesini inceleyin. MySQL veya Oracle veritabanından geçiş yaparken iş sistemlerinizi yeniden yapılandırma maliyetlerini azaltmak için OceanBase iki uyumluluk modunu destekler: Oracle modu ve MySQL modu. Bu modlardaki veri türleri, SQL özellikleri ve iç görünümler MySQL veya Oracle ile tutarlıdır.

Desteklenen özellikler

Ön koşullar

Başlamadan önce aşağıdakilerin sağlandığından emin olun:

• Hedef veritabanı ve tablolar OceanBase'de mevcuttur.

• Bir IP adresi izin listesi yapılandırılmıştır — bkz. İzin listesi grubu yapılandırma

• (CDC kaynak tabloları için) OceanBase Binlog hizmeti etkinleştirilmiştir — bkz. Binlog ile ilgili işlemler

• (Bypass import hedef tabloları için) Bypass import bağlantı noktası etkinleştirilmiştir — bkz. Bypass import

Sınırlamalar

• Ververica Runtime (VVR) 8.0.1 veya sonraki bir sürüm gerektirir.

Anlamsal garantiler:

• CDC kaynak tablolar: tam olarak bir kez semantiği. Tam geçmiş verilerden Binlog okumaya geçiş sırasında, bir arızadan sonra bile veri kaybolmaz veya yinelenmez.

• Hedef tablolar: en az bir kez semantiği. Hedef tabloda bir birincil anahtar varsa, idempotentlik veri doğruluğunu garanti eder.

VVR 11.4.0 CDC mimari değişikliği

VVR 11.4.0 sürümünden itibaren OceanBase CDC konektörü yükseltildi:

• OceanBase LogProxy hizmetine dayalı orijinal CDC konektörü kullanımdan kaldırıldı ve kaldırıldı.

• Artımlı günlük yakalama artık OceanBase Binlog hizmetini gerektirir. OceanBase CDC konektörü, standart MySQL CDC konektörünü doğrudan bağlamaya kıyasla Binlog hizmetiyle daha iyi protokol uyumluluğu ve bağlantı kararlılığı sunar. Değişiklik takibi için standart MySQL CDC konektörünü OceanBase Binlog hizmetine bağlamanız önerilmez.

• Oracle uyumluluk modunda artımlı değişiklik takibi artık desteklenmemektedir. Oracle modu CDC için OceanBase Enterprise Teknik Desteği ile iletişime geçin.

Söz dizimi

CREATE TABLE oceanbase_source (
   order_id     INT,
   order_date   TIMESTAMP(0),
   customer_name STRING,
   price        DECIMAL(10, 5),
   product_id   INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector'  = 'oceanbase',
  'url'        = '<your-jdbc-url>',
  'tableName'  = '<your-table-name>',
  'userName'   = '<your-username>',
  'password'   = '<your-password>'
);

Hedef yazma davranışı:

Gelen her kayıt için konektör, hedef tablonun şemasına göre bir SQL deyimi oluşturur:

• Birincil anahtar yok: INSERT INTO

• Birincil anahtar var: UPSERT (veritabanı uyumluluk moduna bağlı olarak)

WITH parametreleri

Tür eşlemesi

Örnekler

-- OceanBase CDC source table (reads full history, then 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>'
);

-- OceanBase JDBC sink table (for real-time streaming writes)
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>'
);

-- OceanBase bypass import sink table (for high-throughput batch writes)
-- Requires a bounded data source; set Flink to batch mode for best performance
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;

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

-- OceanBase dimension table with ALL cache policy
-- ALL cache loads the full table at startup — suitable for small, stable tables
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;

Sonraki adımlar

• Desteklenen konektörler — Realtime Compute for Apache Flink'te bulunan konektörlerin tam listesi

• OceanBase Binlog hizmeti genel bakış — CDC artımlı okumalar için gereklidir