Oracle实时Doris

本文提供Oracle实时Doris场景的实施思路、配置示例和验证方式，可作为实际项目配置参考。

实现在Oracle中结果表向目标表中实时同步到Doris数据

案例

Oracle中将cdc_source中的数据实时同步到cdc_target中

前提

• 启用Oracle的logminer

解决方案

创建Flink SQL作业类型，添加以下依赖

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

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;

注意

doris数据有多种格式同步数据

csv格式，可能会有换行符问题

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格式，推荐使用JSON格式同步doris数据

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