我在用flinkcdc把数据从sqlserver写到doris

正常情况下sqlserver有删除数据，doris是能捕获到并很快同步删除的。

但是我现在情况是doris做为数仓，数据写到ods，ods的数据还会通过flink计算后写入dwd层，所以此时ods的数据是删除了，但是dwd甚至ads的都没删除，这样就会有脏数据。此时我们就需要去捕获被删除的数据，就要用到debezium插件。

利用cdc把数据从sqlserver发送到kafka，并转为debezium的json格式。

source表正常配，sink-kafka的表配置如下：

然后再拉一个工作流

 

想获取数据的话，就用如下方式：

很多人不懂op是什么，c是什么，接下来解释下： 

debezium会有四种op的数据，

• op ：表示当前事件的类型，取值为：c表示insert、u表示update、d表示delete、r表示快照read；
• ts_ms: connector处理该事件的本地时间戳，可以省略；
• before：变化事件发生之前的值；
• after：变化事件发生之后的值；
• source：事件源的结构信息，包括connector版本、事务ID等；

 举个例子：

• 新增一条数据：

insert into inventory.a values (4, 'n4');

控制台输出的是：

..."payload":{"before":null,"after":{"id":4,"name":"n4"}..."op":"c"...

• 修改一条数据：

update inventory.a set name = 'n4-upd' where id = 4;

控制台输出的是：

..."payload":{"before":{"id":4,"name":"n4"},"after":{"id":4,"name":"n4-upd"}..."op":"u"...

• 删除一条数据：

delete from inventory.a where id = 1;

控制台输出的是：

..."payload":{"before":{"id":1,"name":"n1"},"after":null..."op":"d"...

这样就可以获取到被删除的数据，或者是更新前的数据啦~