大数据之FlinkCDC

最近在做FLinkCDC数据实时同步的数据抽取处理

目标:

将源端系统Oracle数据库的实时数据通过FLINKCDC的形式抽取到Doris中

问题:

在抽取的过程中,如果表的数据量太大,抽取超过30张表以后,所有的任务大概运行25~30分钟以后,所有的任务的状态会从running 变为 Failed.

解决方案:

1.第一解决方案(没有解决掉)

当时通过排查任务发现,我们的Flink部署搭建是通过采用Flink StandAlone HA的模式,有三台服务器,当提交任务到主节点以后,发现主节点上的任务运行大概30分钟的时候,服务器的cpu利用率大概是4250%,导致任务宕机.

所以我们采取的措施是: 将服务器升级,从原本的32核任务扩展到64核,但是升级以后,发现任务运行一段时间以后,还是变成Faild状态

第二种解决方案:(暂时解决掉,生效)

步骤:

1.第一次将全量数据在Dinky中通过JDBC的方式全量抽取过来

2.在启动全量抽取数据的同时,启动FlinkCDC的增量模式,进行增量数据的抽取

具体方案如下:

全量抽取:

create table xxx(

`ID` STRING ,

//建表语句

primary key (ID) not enforced

)

with

(

'connector' = 'jdbc',

'url' = 'jdbc:oracle:thin:@ip:1521/orcl',

'driver' = 'oracle.jdbc.driver.OracleDriver',

'username' = 'xxx',

'password' = ''xxx,

'table-name' = '表名'

);

//Doris建表

create table xxx(

`id` string ,

 //建表语句

primary key (id) not enforced

)

WITH

(

'connector' = 'doris',

'fenodes' = '10.100.XXX:8030',

'table.identifier' = '表名',

'username' = 'root',

'password' = 'xxx,

'sink.properties.format' = 'json',

'sink.properties.read_json_by_line' = 'true',

'sink.label-prefix' = '5410923'

);

insert into xxx

select * from xxxx;

2.增量抽取

create table xxxx (
ID STRING ,
xxxx
primary key (ID) not enforced
)
with
(
'connector' = 'oracle-cdc',
'hostname' = 'xxx',
'port' = '1521',
'username' = 'xxx',
'password' = 'Log#xxx',
'database-name' = 'ORCL',
'schema-name' = 'xxxx',
'table-name' = 'xxxx',
#增量模式
**'scan.startup.mode' = 'latest-offset', **
'debezium.log.mining.strategy' = 'online_catalog',
'scan.incremental.snapshot.chunk.key-column' = 'ID',
'debezium.errors.max.retries' = '3',
'debezium.log.mining.continuous.mine' = 'true',
'debezium.database.tablename.case.insensitive' = 'false'
);

create table xxxx (
id string ,

primary key (`id`) not enforced

)
WITH
(
#同全量语句
);

结果: