2000元阿里云代金券免费领取，2核4G云服务器仅664元/3年，新老用户都有优惠，立即抢购>>>
阿里云采购季（云主机223元/3年）活动入口：请点击进入>>>，
阿里云学生服务器（9.5元/月）购买入口：请点击进入>>>，
有大佬知道在使用flink cdc实现数据同步，如何实现如果服务停止了对数据源表的某个数据进行删除操作，重启服务之后目标表能进行对源表删除的数据进行删除吗？
Flink CDC（Change Data Capture）可以捕获源数据库的变更事件，包括插入、更新和删除操作。当Flink CDC连接到源数据库时，它会监听源数据库的binlog（二进制日志），当源数据库发生变更时，Flink CDC会将这些变更事件发送到Flink Streaming中。
对于你的问题，如果服务停止时源表发生了删除操作，那么这些删除操作对应的变更事件将会被保存在Flink CDC的变更日志中。当服务重启时，Flink CDC会从变更日志中读取这些未处理的变更事件，并将它们发送到Flink Streaming中。因此，如果你的Flink Streaming作业配置了正确的逻辑来处理这些删除事件（例如，使用Table API或DataStream API中的remove()函数），那么它应该能够正确地处理这些删除操作。
具体来说，你可以使用Flink CDC提供的SourceFunction来读取源数据库的变更事件，然后使用DataStream API中的remove()函数来处理这些删除事件。例如：
FlinkSourceConnectorCdc.SourceFunction sourceFunction = ...; // 创建CDC的SourceFunction
DataStream<RowData> changeEvents = ...; // 从sourceFunction获取变更事件
DataStream<RowData> deletedEvents = changeEvents
.filter(new FilterFunction<RowData>() {
@Override
public boolean filter(RowData row) throws Exception {
return row.getRowKind() == RowKind.DELETE;
}
});
deletedEvents.addSink(new SinkFunction() {
@Override
public void invoke(Object value) throws Exception {
// 处理删除事件
}
});
在这个例子中，我们首先创建了一个FlinkSourceConnectorCdc的SourceFunction来读取源数据库的变更事件，然后我们过滤出删除事件，并将它们添加到一个sink中进行处理。
