Realtime Compute for Apache Flink:OceanBase

Informations générales

OceanBase est un système de gestion de base de données natif distribué hybride transactionnel et analytique (HTAP). Pour plus d'informations, consultez le site web d'OceanBase. Afin de réduire les coûts de refonte de vos systèmes métier lors de la migration depuis une base de données MySQL ou Oracle, OceanBase prend en charge deux modes de compatibilité : le mode Oracle et le mode MySQL. Les types de données, les fonctionnalités SQL et les vues internes de ces modes sont cohérents avec ceux de MySQL ou d'Oracle.

Fonctionnalités prises en charge

Prérequis

Avant de commencer, assurez-vous que :

• La base de données cible et les tables existent dans OceanBase.

• Une liste d'autorisation d'adresses IP est configurée. Consultez Configurer un groupe de liste d'autorisation.

• (Pour les tables source CDC) Le service Binlog d'OceanBase est activé. Consultez Opérations liées au Binlog.

• (Pour les tables de destination avec import direct) Le port d'import direct est activé. Consultez Import direct.

Limitations

• Requiert Ververica Runtime (VVR) 8.0.1 ou version ultérieure.

Garanties sémantiques :

• Tables source CDC : sémantique exactly-once. Les données ne sont ni perdues ni dupliquées lors de la transition entre les données historiques complètes et la lecture du Binlog, même après une panne.

• Tables de destination : sémantique at-least-once. Si la table de destination possède une clé primaire, l'idempotence garantit l'exactitude des données.

Modification de l'architecture CDC dans VVR 11.4.0

À partir de VVR 11.4.0, le connecteur CDC OceanBase a été mis à niveau :

• Le connecteur CDC original basé sur le service LogProxy d'OceanBase est obsolète et supprimé.

• La capture des journaux incrémentiels nécessite désormais le service Binlog d'OceanBase. Le connecteur CDC OceanBase offre une meilleure compatibilité protocolaire et une stabilité de connexion supérieure avec le service Binlog par rapport à une connexion directe via le connecteur MySQL CDC standard. Il n'est pas recommandé de connecter le connecteur MySQL CDC standard au service Binlog d'OceanBase pour le suivi des modifications.

• Le suivi des modifications incrémentielles en mode de compatibilité Oracle n'est plus pris en charge. Pour la CDC en mode Oracle, contactez le support technique entreprise d'OceanBase.

Syntaxe

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>'
);

Comportement d'écriture de la destination :

Pour chaque enregistrement entrant, le connecteur construit une instruction SQL basée sur le schéma de la table de destination :

• Aucune clé primaire : INSERT INTO

• Clé primaire présente : UPSERT (selon le mode de compatibilité de la base de données)

Paramètres WITH

Mappage de types

Exemples

-- 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;

Étapes suivantes

• Connecteurs pris en charge — liste complète des connecteurs disponibles dans Realtime Compute for Apache Flink

• Aperçu du service Binlog d'OceanBase — requis pour les lectures incrémentielles CDC