JSON Format

功能描述

JSON Format 能读写 JSON 格式的数据。当前，JSON schema 是从 table schema 中自动推导而得的。

支持的Connector

• Kafka
• Upsert Kafka
• Elasticsearch

参数说明

示例

该示例是从kafka的一个topic中读取数据，并使用kafka sink将数据写入到kafka的另一个topic中。

1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性。若能连通，则表示跨源已经绑定成功；否则表示未成功
2. 创建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;

3. 向作为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"}

4. 读取作为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"}