背景
最近在做一个数据接入的部分事情,从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 message
currentRow.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 kind
currentRow.setRowKind(RowKind.INSERT);
// save to state
state.update(currentRow);
} else {
// DELETE or UPDATER_BEFORE
if (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 removed
state.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
版权归原作者 鸿乃江边鸟 所有, 如有侵权,请联系我们删除。