Flink：FlinkSql解析嵌套Json-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink：FlinkSql解析嵌套Json。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

日常开发中都是用的简便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' -- 解析失败跳过
)

2,数据格式以及对应的flinksql格式 Flink：FlinkSql解析嵌套Json