本文主要是介绍用flinkcdc debezium来捕获数据库的删除内容，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
我在用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"...这样就可以获取到被删除的数据，或者是更新前的数据啦~
这篇关于用flinkcdc debezium来捕获数据库的删除内容的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
