日常开发中都是用的简便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格式

 

引自：https://blog.csdn.net/qq_21383435/article/details/124889251