Oracle real-time Doris

This article provides implementation ideas, configuration examples and verification methods for Oracle real-time Doris scenarios, which can be used as a reference for actual project configuration.

Real-time synchronization of the result table in Oracle to the target table to Doris data

Case

Synchronize data in cdc_source to cdc_target in real time in Oracle

Premise

• Enable Oracle's logminer

solution

Create a Flink SQL job type and add the following dependencies

• flink-sql-connector-oracle-cdc-3.2.1.jar download
• flink-doris-connector-1.18-25.0.0.jar download

CREATE TABLE from_table
(
    USERNAME STRING,
    AGE      INT,
    PRIMARY KEY (USERNAME) NOT ENFORCED
) WITH (
      'connector' = 'oracle-cdc',
      'hostname' = 'localhost',
      'port' = '30110',
      'username' = 'root',
      'password' = 'root123',
      'database-name' = 'PUBLIC',
      'schema-name' = 'root',
      'table-name' = 'CDC_SOURCE',
      'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE target_table
(
    USERNAME STRING,
    AGE      INT,
    PRIMARY KEY (USERNAME) NOT ENFORCED
) WITH (
      'connector' = 'doris',
      'fenodes' = 'localhost:8030',
      'table.identifier' = 'ispong_db.cdc_target',
      'username' = 'root',
      'password' = 'root123')
;

INSERT INTO target_table (USERNAME, AGE)
select USERNAME, AGE
from from_table;

Note

Doris data has multiple formats to synchronize data

csv format, there may be line break problems

CREATE TABLE target_table
(
    USERNAME STRING,
    AGE      INT,
    PRIMARY KEY (USERNAME) NOT ENFORCED
) WITH (
      'connector' = 'doris',
      'fenodes' = 'localhost:8030',
      'table.identifier' = 'ispong_db.cdc_target',
      'username' = 'root',
      'password' = 'root123',
      'sink.properties.format' = 'csv',
      'sink.properties.column_separator' = '\t',
      'sink.properties.row_delimiter' = '\n', 
      'sink.properties.quote' = '"',        
      'sink.max-retries' = '3'
    );

JSON format, it is recommended to use JSON format to synchronize doris data

CREATE TABLE target_table
(
    USERNAME STRING,
    AGE      INT,
    PRIMARY KEY (USERNAME) NOT ENFORCED
) WITH (
      'connector' = 'doris',
      'fenodes' = 'localhost:8030',
      'table.identifier' = 'ispong_db.cdc_target',
      'username' = 'root',
      'password' = 'root123',
      'sink.properties.format' = 'json',
      'sink.properties.read_json_by_line' = 'true'
    );