Flink：FlinkSql解析嵌套Json

阿里云国内75折回扣微信号：monov8

阿里云国际，腾讯云国际，低至75折。AWS 93折免费开户实名账号代冲值优惠多多微信号：monov8 飞机：@monov6

日常开发中都是用的简便json格式但是偶尔也会遇到嵌套json的时候因此在用flinksql的时候就有点麻烦下面用简单例子简单定义处理下

1数据是网上摘抄但包含里常用的大部分格式

{
   "afterColumns": {
       "created": "1589186680",
       "extra": {
           "canGiving": false
       },
       "parameter": [1, 2, 3, 4]
   },
   "beforeColumns": null,
   "tableVersion": {
       "binlogFile": null,
       "binlogPosition": 0,
       "version": 0
   },
   "touchTime": 1589186680591,
"type":3,
   "arr": [{
       "address": "北京市海淀区",
       "city": "beijing"
   }, {
       "address": "北京市海淀区",
       "city": "beijing"
   }, {
       "address": "北京市海淀区",
       "city": "beijing"
   }]
}

注意

Json 中的每个 {} 都需要用 Row 类型来表示
Json 中的每个 [] 都需要用 Arrary 类型来表示
数组的下标是从 1 开始的不是 0
查询select时关键字需要加反引号如上面 SQL 中的 `type`
select 语句中的字段类型和顺序一定要和结果表的字段类型和顺序保持一致

因此FlinkSql语句应该为

CREATE TABLE kafka_source (
afterColumns ROW(created STRING,extra ROW(canGiving BOOLEAN),`parameter` ARRAY <INT>) ,
beforeColumns STRING ,
tableVersion ROW(binlogFile STRING,binlogPosition INT ,version INT) ,
touchTime bigint,
`type` INT,
arr ARRAY<ROW<address STRING,city STRING>>
) WITH (
'connector' = 'kafka', -- 使用 kafka connector
'topic' = 'test', -- kafka topic
'properties.bootstrap.servers' = 'master:9092,storm1:9092,storm2:9092', -- broker连接信息
'properties.group.id' = 'jason_flink_test', -- 消费kafka的group_id
'scan.startup.mode' = 'latest-offset', -- 读取数据的位置
'format' = 'json', -- 数据源格式为 json
'json.fail-on-missing-field' = 'true', -- 字段丢失任务不失败
'json.ignore-parse-errors' = 'false' -- 解析失败跳过
)