FLinkCDC引起的生产事故（二）

news2024/9/26 5:22:15

背景：

最近在做实时数据的抽取工作，利用FLinkCDC实时抽取目标库Oracle的数据到Doris中，但是在抽取的过程中，会导致目标库的生产库数据库非常卡顿，为了避免对生产环境的数据库造成影响，对生产环境的数据库利用OGG技术做了备库，从备库中利用FlinkCDC抽取数据到Doris，但是在抽取的过程中出现了同样的错误，导致备库数据库卡顿，数据异常，数据库宕机，至于这些原因，怀疑我们当时的方案出了严重的问题

第一次发方案：

直接利用FLINKCDC抽取Oralce的binglog日志，这种方案就是在Dinky中启动的每一个任务都会去Oracle的源端数据库中读取binglog日志，从而有大量的进程和线程出现，导致cpu和内存无限上升（刚开始cpu是8核，后面升级到了32核，问题还是出现）

架构图如下：

改进后的方案：

Oracle CDC数据表主要用于获取Oracle 数据，并可以实时同步数据表中的修改，经常用在复杂的计算场景。例如，作为一张维表和其他数据表做Join操作。在使用中，同一张MySQL表可能被多个作业依赖，当多个任务使用同一张MySQL表做处理时，MySQL数据库会启动多个连接，对MySQL服务器和网络造成很大的压力。

为了缓解对上游Oracle 数据库的压力，Flink实时计算已提供Oracle 整库同步到Kafka的能力，通过引入Kafka作为中间层，利用OGG将数据推送到Kafka，然后FLink从Kafka获取数据，这样减少了源端数据库的压力

架构图如下：

基本操作如下：

CREATE TEMPORARY TABLE tempOrder (
  `key_order_id` BIGINT NOT NULL,
  `value_product` STRING,
  PRIMARY KEY (key_order_id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'order',
  'properties.bootstrap.servers' = 'xxxx',
  'key.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.format' = 'json',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY',
  'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
  'value.json.infer-schema.primitive-as-string' = 'false'
);