Re:flink sql解析kafka数据

Wed, 06 Jul 2022 01:20:18 -0700 

Hi,


你可以在定义表ccc_test_20220630_2字段的时候，结构如果固定，可以指定字段类型为ARRAY+ROW吧，例如 abc 
ARRAY<ROW<key1 STRING, key2 STRING>>，如果里面是动态结构，可以定义为MAP
结构如果比较复杂，或者字段不明确，就自定义UDF解决。


Thanks




在 2022-06-30 15:02:55，"小昌同学" <ccc0606fight...@163.com> 写道：

各位大佬  请教一下就是我kafka的数据是这样的嵌套格式 ARRAY里面嵌套了ROW类型  我这边想直接通过flink sql建表语句拿到最里面的字段的值
我百度找到了 'json.infer-schema.flatten-nested-columns.enable'='true'  但是我在客户端执行的时候  
发现识别不到这个字段  
有大佬遇到我这样的问题嘛 或者有啥其他的解决法子嘛
CREATE TABLE ccc_test_20220630_2
(  
               trans_number   STRING,
               end_timestamp  STRING,
               return_flag    STRING,
               commodity_type STRING

   
)
   COMMENT '中台交易流水小票头' WITH (
   'connector' = 'kafka',
   'topic' = 'yh_rme_soc_stream_prod-tlog_header',
   'properties.bootstrap.servers' = '****',
   'properties.group.id' = 'ccc_test_20220630_2',
   'properties.request.timeout.ms' = '60000',
   'format' = 'json',
   'scan.startup.mode' = 'group-offsets',
   -- 'scan.startup.mode' = 'timestamp',
   -- 'scan.startup.timestamp-millis' = '1653739200000',
   'json.fail-on-missing-field' = 'false',
   'json.ignore-parse-errors' = 'true'
'json.infer-schema.flatten-nested-columns.enable'='true'
);


| |
小昌
|
|
ccc0606fight...@163.com
|

回复