Kafka JSON格式转换

{
 	"funcName": "test",
 	"data": {
  		"snapshots": [{
  			"content_type": "application/x-gzip-compressed-jpeg",
   			"url": "https://blog.csdn.net/xianpanjia4616"
  			 }],
 	 "audio": [{
   			"content_type": "audio/wav",
   			"url": " https://bss.csdn.net/m/topic/blog_star2020/detail?username=xianpanjia4616"
	 	 	}]
 		},
	"resultMap": {
  		"result": {
  			"cover": "/data/test/log.txt"
    		},
  		"isSuccess": true
 		},
 	"meta": {
  		"video_type": "normal"
 	},
 	"type": 2,
 	"timestamp": 1610549997263,
 	"arr": [{
  		"address": "北京市海淀区",
  		"city": "beijing"
 		}, {
  		"address": "北京市海淀区",
  		"city": "beijing"
 		}, {
  		"address": "北京市海淀区",
 	 	"city": "beijing"
 		}],
 	"map": {
  		"flink": 456
 		},
 	"doublemap": {
  		"inner_map": {
   			"key": 123
  			}
 		}
}

CREATE TABLE kafka_source (
    funcName STRING,
    data ROW>,audio ARRAY>>,
    resultMap ROW<`result` MAP,isSuccess BOOLEAN>,
    meta  MAP,
    `type` INT,
    `timestamp` BIGINT,
    arr ARRAY>,
    map MAP,
    doublemap MAP>,
    proctime as PROCTIME()
) WITH ()

select
funcName,
doublemap['inner_map']['key'],
count(data.snapshots[1].url),
`type`,
TUMBLE_START(proctime, INTERVAL '30' second) as t_start
from kafka_source
group by TUMBLE(proctime, INTERVAL '30' second),funcName,`type`,doublemap['inner_map']['key']

1.Json 中的每个 {} 都需要用 Row 类型来表示
2.Json 中的每个 [] 都需要用 Arrary 类型来表示
3.数组的下标是从 1 开始的不是 0 如上面 SQL 中的 data.snapshots[1].url
4.关键字在任何地方都需要加反引号 如上面 SQL 中的 type