Flink Format系列(1)-JSON

Flink的json格式支持读和写json格式的数据，只需要指定 'format' = 'json'，下面以kafka为例。

CREATE TABLE teacher (
  name STRING,
  age INT,
  title ROW<id INT, name STRING, level INT>,
  addr STRING
) WITH (
 'connector' = 'kafka',
 'topic' = 'test_teacher_info',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'json',
 'json.fail-on-missing-field' = 'false',
 'json.ignore-parse-errors' = 'true'
)

关键配置的说明：

json.fail-on-missing-field：就是在json缺失字段时是否报错

json.ignore-parse-errors：在解析json失败时是否报错

一般无法保证json格式，所以以上两个配置是比较重要的。

因为有ROW和ARRAY、MAP的数据类型，所以可以解析复杂的嵌套的json格式，通过xx.xx来访问内部嵌套的json字段。比如上面的就可以通过 select title.name 来访问title这个子json里的name值。比如来的数据是：

{"name":"jack", "age":44, "title":{"id":1, "name":"副教授", "level": 3}, "addr":"xxxx1"}

{"name":"rose", "age":47, "title":{"id":1, "name":"副教授", "level": 2}, "addr":"xxxx2"}

这里注意title的嵌套json的格式，别写成了 "title":\"{\"id\":1, \"name\":\"副教授\", \"level\": 3}\"这种格式，不然会报错，无法将TextNode转化为ObjectNode。

一些使用注意说明：

1）Json 中的每个 {} 都需要用 Row 类型来表示
2）Json 中的每个 [] 都需要用 Arrary 类型来表示
3）数组的下标是从 1 开始的不是 0
4）SQL中字段名是关键字时，需要用反引号修饰 例如：`timestamp`
5）select 语句中的字段类型和顺序一定要和结果表的字段类型和顺序保持一致

json类型和flink sql类型的对应关系