JSON Format
功能描述
JSON Format 能读写 JSON 格式的数据。当前,JSON schema 是从 table schema 中自动推导而得的。
支持的Connector
- Kafka
- Upsert Kafka
- Elasticsearch
参数说明
参数 |
是否必选 |
默认值 |
类型 |
说明 |
---|---|---|---|---|
format |
是 |
(none) |
String |
声明使用的格式,这里应为'json'。 |
json.fail-on-missing-field |
否 |
false |
Boolean |
当解析字段缺失时,是跳过当前字段或行,还是抛出错误失败(默认为 false,不抛出错误失败)。 |
json.ignore-parse-errors |
否 |
false |
Boolean |
当解析异常时,是跳过当前字段或行,还是抛出错误失败(默认为 false,即抛出错误失败)。如果忽略字段的解析异常,则会将该字段值设置为null。 |
json.timestamp-format.standard |
否 |
'SQL' |
String |
声明输入和输出的TIMESTAMP和TIMESTAMP WITH LOCAL TIME ZONE 的格式。 当前支持的格式为'SQL'和'ISO-8601':
|
json.map-null-key.mode |
否 |
'FALL' |
String |
指定处理 Map 中 key 值为空的方法。当前支持的值有:'FAIL','DROP'和'LITERAL'。
|
json.map-null-key.literal |
否 |
'null' |
String |
当 'json.map-null-key.mode' 是LITERAL的时候,指定字符串常量替换 Map 中的空 key 值。 |
示例
该示例是从kafka的一个topic中读取数据,并使用kafka sink将数据写入到kafka的另一个topic中。
- 根据kafka所在的虚拟私有云和子网创建相应的跨源,并绑定所要使用的队列。然后设置安全组入向规则,使其对当前将要使用的队列放开,并根据kafka的地址测试队列连通性。若能连通,则表示跨源已经绑定成功;否则表示未成功
- 创建flink opensource sql作业,并选择flink版本为1.12,选择保存日志,然后提交并运行,其SQL代码如下:
CREATE TABLE kafkaSource ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = '<yourSourceTopic>', 'properties.bootstrap.servers' = '<yourKafkaAddress>:<yourKafkaPort>', 'properties.group.id' = '<yourGroupId>', 'scan.startup.mode' = 'latest-offset', "format" = "json" ); CREATE TABLE kafkaSink ( order_id string, order_channel string, order_time string, pay_amount double, real_pay double, pay_time string, user_id string, user_name string, area_id string ) WITH ( 'connector' = 'kafka', 'topic' = '<yourSinkTopic>', 'properties.bootstrap.servers' = '<yourKafkaAddress>:<yourKafkaPort>', "format" = "json" ); insert into kafkaSink select * from kafkaSource;
- 向作为source的kafka的topic中插入下列数据:
{"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"} {"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}
- 读取作为sink的kafka的topic中的数据,其结果如下:
{"order_id":"202103241000000001","order_channel":"webShop","order_time":"2021-03-24 10:00:00","pay_amount":100.0,"real_pay":100.0,"pay_time":"2021-03-24 10:02:03","user_id":"0001","user_name":"Alice","area_id":"330106"} {"order_id":"202103241606060001","order_channel":"appShop","order_time":"2021-03-24 16:06:06","pay_amount":200.0,"real_pay":180.0,"pay_time":"2021-03-24 16:10:06","user_id":"0001","user_name":"Alice","area_id":"330106"}