记一次 Flink mongoDB CDC 到Kafka遇到的问题

背景

最近在做一个数据接入的部分事情，从mongo导入到 adb，趁着做的事情聊一下Flink内部的一些机制。
首先这会拆分两个部分，一部分是从 mongo 到 Kafka，另一部分是从 Kafka 到 adb，其中遇到了一些问题，比如说 CDC 的机制，
upset kafka source 和 kafka source的一些区别等
mongo 的版本为 4.4.x

分析

mongo -> kafka

一开始时候 Flink source 是 mongo cdc sink 选择是 正常的 kafka
部分配置如下:

// source
CREATE TABLE products (...PRIMARY KEY(_id) NOT ENFORCED
) WITH ('connector' = 'mongodb-cdc','hosts' = 'localhost:27017,localhost:27018,localhost:27019','username' = 'flinkuser','password' = 'flinkpw','database' = 'inventory','collection' = 'products'
);// sink
CREATE TABLE KafkaTable (`ts` TIMESTAMP(3) METADATA FROM 'timestamp',`user_id` BIGINT,`item_id` BIGINT,`behavior` STRING
) WITH (..'connector' = 'kafka','key.json.ignore-parse-errors' = 'true','format' = 'debezium-json',
)

这里选择的format是 debezium-json ,这在后续读取kafka数据进行 Row Number over操作的时候，会报错：

StreamPhysicalOverAggregate doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[]..)

从意思来看 Row Number over 操作是不支持 CDC产生的数据的(CDC会产生 +i +U -U 等数据)，于是选择了 upsert kafka,upsert kafka这里会有一个解释:

作为 sink，upsert-kafka 连接器可以消费 changelog 流。它会将 INSERT/UPDATE_AFTER 数据作为正常的 Kafka 消息写入，并将 DELETE 数据以 value 为空的 Kafka 消息写入（表示对应 key 的消息被删除）

所以这里我们选择把 kakfa的数据转换成的正常的 数据流，而不是CDC数据，因为我们最终存储的 Adb 是可以支持upsert操作。
可以看到Flink 物理计划中 会额外多出一个StreamExecChangelogNormalize算子,该流的具体流如下：

mongo(4.4.x) ->  StreamExecChangelogNormalize -> ConstraintEnforcer(NotNullEnforcer(fields=[_id])) -> kafkaSink

可以看到StreamExecChangelogNormalize是在kafka sink之前的，也就是说StreamExecChangelogNormalize是用来Flink用来产生CDC数据的,Flink SQL Planner 会自动为 Upsert 类型的 Source 生成一个 ChangelogNormalize 节点，并按照上述操作将其转换为完整的变更流；代价则是该算子节点需要存储体积巨大的 State 数据。具体可以参考深入解读 MongoDB CDC 的设计与实现,产生的CDC数据流如下:

StreamExecChangelogNormalize.translateToPlanInternal||\/ProcTimeDeduplicateKeepLastRowFunction.processElement||\/ProcTimeDeduplicateKeepLastRowFunction.processLastRowOnChangelog 

processLastRowOnChangelog这里会存有 keyedState 状态，但是为了补足这个带有CDC的数据的，所以这里得有依赖状态在flink端进行状态的转换，具体可以看：DeduplicateFunctionHelper.processLastRowOnChangelog 方法：

 static void processLastRowOnChangelog(RowData currentRow,boolean generateUpdateBefore,ValueState<RowData> state,Collector<RowData> out,boolean isStateTtlEnabled,RecordEqualiser equaliser)throws Exception {RowData preRow = state.value();RowKind currentKind = currentRow.getRowKind();if (currentKind == RowKind.INSERT || currentKind == RowKind.UPDATE_AFTER) {if (preRow == null) {// the first row, send INSERT messagecurrentRow.setRowKind(RowKind.INSERT);out.collect(currentRow);} else {if (!isStateTtlEnabled && equaliser.equals(preRow, currentRow)) {// currentRow is the same as preRow and state cleaning is not enabled.// We do not emit retraction and update message.// If state cleaning is enabled, we have to emit messages to prevent too early// state eviction of downstream operators.return;} else {if (generateUpdateBefore) {preRow.setRowKind(RowKind.UPDATE_BEFORE);out.collect(preRow);}currentRow.setRowKind(RowKind.UPDATE_AFTER);out.collect(currentRow);}}// normalize row kindcurrentRow.setRowKind(RowKind.INSERT);// save to statestate.update(currentRow);} else {// DELETE or UPDATER_BEFOREif (preRow != null) {// always set to DELETE because this row has been removed// even the input is UPDATE_BEFORE, there may no UPDATE_AFTER after it.preRow.setRowKind(RowKind.DELETE);// output the preRow instead of currentRow,// because preRow always contains the full content.// currentRow may only contain key parts (e.g. Kafka tombstone records).out.collect(preRow);// clear state as the row has been removedstate.clear();}// nothing to do if removing a non-existed row}}

kafka -> adb

一开始时候的kafka 我们选择了常规的 kafka source

CREATE TABLE KafkaTable (...
) WITH ('connector' = 'kafka','topic' = 'user_behavior','properties.bootstrap.servers' = 'localhost:9092','properties.group.id' = 'testGroup','scan.startup.mode' = 'earliest-offset','format' = 'json'
)

这里获取到的数据就是 正常的json数据，而不是 debezium-json数据，具体区别，可以参考下面的说明。
如果选择一开始的 kafka sink是 kafka的话 可以看到这里的物理计划流向为:

kafka -> SinkMaterializer -> adb sink 

对于为什么会出现 SinkMaterializer, 为了解决 changlog的乱序问题，为下游提供一个正确的upsert视图, 产生 SinkMaterializer 物理算子的数据流如下:

StreamExecSink ||\/
createSinkTransformation // 这里有   final boolean needMaterialization = !inputInsertOnly && upsertMaterialize; 会插入SinkUpsertMaterializer算子||\/
SinkUpsertMaterializer //table.exec.state.ttl的设置||\/SinkUpsertMaterializer.processElement // 这里有 keyed state

当然SinkUpsertMaterializer 这个算子也是可以通过配置 table.exec.sink.upsert-materialize 控制的

，因为我们现在选择 kafka sink的是upsert kafka 这里会消除掉cdc数据，所以不存在以上的SinkUpsertMaterializer.

debezium-json的格式与 json的格式区别

debezium-json 变成 {before:, after:, op:} before 和after里才是真正的数据
json 直接就是 json格式数据

这里的区别的具体数据流可以参考如下,主要是 对mongo数据的处理：

KafkaDynamicTableFactory.createDynamicTableSink||\/KafkaDynamicSink.getSinkRuntimeProvider||\/final SerializationSchema<RowData> valueSerialization =createSerialization(context, valueEncodingFormat, valueProjection, null);||\/
DynamicKafkaRecordSerializationSchema 这里会用到||\/
valueSerialized = valueSerialization.serialize(valueRow);

valueSerialization 这会有多种序列化的方式,如:
debezium-json 对应 DebeziumJsonSerializationSchema
json对应JsonRowDataSerializationSchema