开发者社区> 问答> 正文

flink 1.11 cdc: kafka中存了canal-json格式的多张表信息,需要按表解析做

mysql表: CREATE TABLE test ( id int(11) NOT NULL, name varchar(255) NOT NULL, time datetime NOT NULL, status int(11) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

CREATE TABLE status ( id int(11) NOT NULL, name varchar(255) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8

kafka中数据: // 表test 中insert事件 {"data":[{"id":"1745","name":"jindy1745","time":"2020-07-03 18:04:22","status":"0"}],"database":"ai_audio_lyric_task","es":1594968168000,"id":42,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)","time":"datetime","status":"int(11)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12,"time":93,"status":4},"table":"test","ts":1594968168789,"type":"INSERT"}

//表status 中的事件 {"data":[{"id":"10","name":"status"}],"database":"ai_audio_lyric_task","es":1595305259000,"id":589240,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12},"table":"status","ts":1595305259386,"type":"INSERT"}

如何由于kafka中的json动态的变化的,比如新增一个表,如何能转成应对的RowData, 感觉无法直接用JsonRowDeserializationSchema或CanalJsonDeserializationSchema来做处理。

*来自志愿者整理的flink邮件归档

展开
收起
小阿矿 2021-12-06 16:47:54 1318 0
1 条回答
写回答
取消 提交回答
  • http://apache-flink.147419.n8.nabble.com/flink-1-10-sql-kafka-format-json-schema-json-object-tt4665.html

    这个邮件里提到了类似的问题。

    https://issues.apache.org/jira/browse/FLINK-18002 这个issue完成后(1.12),你可以将

    “data”,“mysqlType”等格式不确定的字段定义为String类型,

    下游通过udf自己再解析对应的json*来自志愿者整理的flink邮件归档

    2021-12-06 18:28:36
    赞同 展开评论 打赏
问答排行榜
最热
最新

相关电子书

更多
Java Spring Boot开发实战系列课程【第16讲】:Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析 立即下载
MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute 立即下载
消息队列kafka介绍 立即下载

相关镜像