数据流图是 mongodb --> flink cdc --> kafka (canal json)
看了flink cdc解析出的mongodb oplog转成json字符串是下面这样子[1]，而下游需要从kafka消费canal
json格式的消息，中间的格式转换得自己实现是么？
但mongodb oplog是不带schema信息的，而且没有canal中的old字段信息，这块信息要怎么转换呢？
另，我用flink sql如下往kafka发送canal json格式数据是不完整的[2]，并不是一个标准的canal
json数据[3]。这是已知的issue么？
CREATETABLE
mongo_customers(_idSTRING,customer_idINT,nameSTRING,addressSTRING,PRIMARYKEY(_id)NOTENFORCED)WITH('connector'='mongodb-cdc','hosts'='localhost:27017','username'='mongouser','password'='mongopw','database'='mgdb','collection'='customers');
CREATETABLE
kafka_customers(_idSTRING,customer_idINT,nameSTRING,addressSTRING,PRIMARYKEY(_id)NOTENFORCED)WITH('connector'='upsert-kafka','topic'='customers','properties.bootstrap.servers'='localhost:9092',
'format'='canal-json');
INSERT INTO kafka_customers SELECT * FROM mongo_customers;
[1]
{
"_id": "{\"_id\": {\"$oid\": \"614a9ae069736f5fcc48fe59\"}, \"copyingData\":
true}",
"operationType": "insert",
"fullDocument": "{\"_id\": {\"$oid\": \"614a9ae069736f5fcc48fe59\"},
\"_class\": \"com.huifu.uqp.dal.mongo.model.TopTransOrder\", \"reqSeqId\":
\"123\", \"ordId\": \"1440509760709632000\", \"outTransId\": \"123\",
\"merOrdId\": \"123\", \"hfSeqId\": \"123\", \"partyOrderId\": \"123\",
\"bankSeqId\": \"123\", \"orgOrdId\": \"123\", \"orgTermOrdId\": \"123\",
\"orgHuifuSeqId\": \"123\", \"transDate\": \"20210913\", \"productId\":
\"app8\", \"serviceId\": \"6767639\", \"topAgentId\": \"123\",
\"belongAgentId\": \"123\", \"chainsId\": \"123\", \"huifuId\":
\"666684552350\", \"transMajorCategory\": \"123\", \"consoleActualPayChannel\":
\"123\", \"consolePayType\": \"123\", \"consolePreAuthFlag\": \"123\",
\"consoleSubsidyFlag\": \"123\", \"consoleDcType\": \"123\", \"consoleIsFq\":
\"123\", \"consoleAcctDivFlag\": \"123\", \"actualPayChannel\": \"123\",
\"payChannel\": \"123\", \"transType\": \"123\", \"payType\": \"123\",
\"dcType\": \"123\", \"isAcctDiv\": \"123\", \"isDelayAcct\": \"123\",
\"creditType\": \"123\", \"devsId\": \"123\", \"ordAmt\": 123.32, \"feeAmt\":
123.0, \"actOrdAmt\": 123.0, \"actualRefAmt\": 123.0, \"refAmt\": 123.0,
\"refFeeAmt\": 123.0, \"subsidyAmt\": 123.0, \"subsidyRefAmt\": 123.0,
\"payCardId\": \"123\", \"feeRecType\": \"123\", \"feeFlag\": \"123\",
\"transStat\": \"S\", \"createTime\": {\"$date\": 1632279264987},
\"transFinishTime\": \"123\", \"kafkaTime\": \"123\", \"tableName\": \"123\",
\"offset\": \"123\", \"recordVersion\": \"123\", \"sign\": \"123\"}",
"source": {
"ts_ms": 0,
"snapshot": "true"
},
"ns": {
"db": "amp_test",
"coll": "TopTransOrder"
},
"to": null,
"documentKey": "{\"_id\": {\"$oid\": \"614a9ae069736f5fcc48fe59\"}}",
"updateDescription": null,
"clusterTime": null,
"txnNumber": null,
"lsid": null
}
[2]
{"data":[{"_id":"614a9b3769736f5fcc492613","id":null,"reqSeqId":"123","ordId":"1440510124339011584","outTransId":"123","merOrdId":"123","hfSeqId":"123","partyOrderId":"123","bankSeqId":"123","orgOrdId":"123","orgTermOrdId":"123","orgHuifuSeqId":"123","transDate":"20210913","productId":"app3","serviceId":"6767679","topAgentId":"123","belongAgentId":"123","chainsId":"123","huifuId":"66668455531","transMajorCategory":"123","consoleActualPayChannel":"123","consolePayType":"123","consolePreAuthFlag":"123","consoleSubsidyFlag":"123","consoleDcType":"123","consoleAcctDivFlag":"123","actualPayChannel":"123","payChannel":"123","transType":"123","payType":"123","dcType":"123","isAcctDiv":"123","isDelayAcct":"123","creditType":"123","devsId":"123","ordAmt":123.32,"feeAmt":123,"actOrdAmt":123,"actualRefAmt":123,"refAmt":123,"refFeeAmt":123,"subsidyAmt":123,"subsidyRefAmt":123,"payCardId":"123","feeRecType":"123","feeFlag":"123","transStat":"S","transFinishTime":"123","tableName":"123","offset":"123","recordVersion":"123","sign":"123","synModifyTime":null,"merName":null,"bankName":null,"bankRespDesc":null,"bagentId":null,"bankId":null,"cashRespDesc":null,"reqDate":null,"accSplitBunch":null,"acctId":null,"fqFeeAmt":null,"payCardIdEnc":null,"goodsDesc":null,"remark":null,"synTtlDate":null,"outOrdId":null,"devType":null,"feeHuifuId":null,"feeAcctId":null,"orgTransDate":null,"orgOrdAmt":null,"orgCreateTime":null,"userType":null,"userId":null,"userIdExt":null,"settleAmt":null,"refCnt":null,"consoleCountSum":null,"topConsolePayType":null,"orgMerOrdId":null,"feeAllowanceFlag":null,"correctStat":null,"addedOrgFeeAmt":null,"discountFeeAmt":null,"acctFinishTime":null,"pospSeqId":null,"outOrderId":null,"cashTransId":null,"orgPayType":null,"orgPayChannel":null,"branch1HuifuId":null,"branch2HuifuId":null,"branch3HuifuId":null,"branch4HuifuId":null,"branch5HuifuId":null,"branchHuifuId":null,"level":null,"branchChannelId":null,"orgFeeAmt":null,"orgConsoleIsFq":null,"orgCreditType":null,"fqMerDiscountFlag":null,"payScene":null,"labels":null,"orgTransType":null,"orgFeeRecType":null,"orgFeeFlag":null,"orgDiscountFeeAmt":null,"merOperId":null,"operType":null,"batchId":null,"authNo":null,"refNum":null,"bankMerId":null,"bankMerName":null,"posMerId":null,"posMerName":null,"acqrInstId":null,"doubleExempt":null,"pnrDevId":null,"posTermId":null,"realPayType":null,"channelFinishTime":null,"transRefundBankId":null,"transRefundBankName":null,"orgRealPayType":null,"orgDevsId":null,"merPriv":null,"transRefundOutOrdId":null,"orgHfSeqId":null,"synMode":null,"cloudPay":null,"terminalReqDate":null,"terminalPayChannel":null,"huifuFstOrg":null,"huifuSecOrg":null,"huifuThdOrg":null,"huifuForOrg":null,"huifuSales":null,"partnerBd":null,"organizationId":null,"upperOrgId":null,"merOrg":null,"partnerInnerFstOrg":null,"partnerInnerSecOrg":null,"partnerInnerThdOrg":null,"partnerFstOrg":null,"partnerSecOrg":null,"partnerThdOrg":null,"collectMerFstOrg":null,"collectMerSecOrg":null,"collectMerThdOrg":null,"collectMerForOrg":null,"collectMerFivOrg":null,"collectMerSixOrg":null,"fullPath":null}],"type":"INSERT"}
[3]
https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/table/formats/canal/#%e5%a6%82%e4%bd%95%e4%bd%bf%e7%94%a8-canal-format
在 2022-08-22 22:57:04，"Xuyang" <xyzhong...@163.com> 写道：
>Hi, 请问你的需求是 “debezium数据”-> flink ->“canal ”么？
>如果是这样的话，可以用UDF[1]来尝试下。<br/><br/>[1]
>https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/udfs/
>在 2022-08-21 10:49:29，"casel.chen" <casel_c...@126.com> 写道：
>>flink cdc如何将捕获变更记录转成canal json格式输出到下游kafka?
>>flink cdc获取的是debezium格式记录（用的是 JsonDebeziumDeserializationSchema)，要如何转换成canal
>>json格式输出呢？有没有例子或关键代码展示？谢谢！
