Flink 1.10-SQL解析复杂json问题

kafka数据写入kafka 数据，flink1.10-sql解析复杂json中字段为string，导致数据丢失。

语句：
CREATE TABLE A (
w_data  STRING,
w_table  STRING,
w_ts TIMESTAMP(3)


CREATE TABLE B (
w_ts TIMESTAMP(3),
city1_id  STRING,
cate3_id  STRING,
pay_order_id  STRING
)

insert into B
select w_ts,

'test' as city1_id,

ArrayIndexOf(w_data, 0) AS cate3_id,
w_data as pay_order_id
from A

部分数据
A
{"w_es":1589870637000,"w_type":"INSERT","w_isDdl":false,"w_data":[{"pay_info":"channelId=82&onlineFee=89.0&outTradeNo=0&payId=0&payType=02&rechargeId=4&totalFee=89.0&tradeStatus=success&userId=32590183789575&sign=00","online_fee":"89.0","sign":"00","account_pay_fee":"0.0"}],"w_ts":"2020-05-20T13:58:37.131Z","w_table":"cccc111"}

B
{"w_ts":"2020-05-20T13:58:37.131Z","city1_id":"test","cate3_id":null,"pay_order_id":""}









[hidden email]