# Maxwell Format
#### 功能描述
Flink 支持将 Maxwell JSON 消息解释为 INSERT/UPDATE/DELETE 消息到 Flink SQL 系统中。在许多情况下，这对于利用此功能很有用。
例如：
- 将数据库中的增量数据同步到其他系统
- 审计日志
- 数据库的实时物化视图
- 临时连接更改数据库表的历史等等。
Flink 还支持将 Flink SQL 中的 INSERT/UPDATE/DELETE 消息编码为 Maxwell JSON 消息，并发送到 Kafka 等外部系统。但是，目前 Flink 无法将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此，Flink 将 UPDATE_BEFORE 和 UDPATE_AFTER 编码为 DELETE 和 INSERT Maxwell 消息。
#### 参数说明
| 参数                                     | 是否必选 | 默认值    | 类型      | 说明                                                                                                                                                                                                                                      |
|:---|:---|:---|:---|:---|
| format                                 | 是    | (none) | String  | 指定使用格式，此处使用'maxwell-json'。                                                                                                                                                                                                              |
| maxwell-json.ignore-parse-errors       | 否    | false  | Boolean | 跳过解析错误而不是失败的字段和行。出现错误时，字段设置为空。                                                                                                                                                                                                          |
| maxwell-json.timestamp-format.standard | 否    | 'SQL'  | String  | 指定输入和输出时间戳格式。当前支持的值为"SQL"和"ISO-8601"：选项"SQL"将以"yyyy-MM-dd HH:mm:ss.s{precision}"格式解析输入时间戳，例如"2020-12-30 12" :13:14.123' 并以相同格式输出时间戳。选项'ISO-8601'将以"yyyy-MM-ddTHH:mm:ss.s{precision}"格式解析输入时间戳，例如'2020-12-30T12： 13:14.123' 并以相同格式输出时间戳。 |
| maxwell-json.map-null-key.mode         | 否    | 'FAIL' | String  | 在序列化地图数据的空键时指定处理模式。当前支持的值为"FAIL"、"DROP"和"LITERAL"：选项"FAIL"将在遇到带有空键的地图时抛出异常。选项"DROP"将删除地图数据的空键条目。选项"LITERAL"将替换空带字符串文字的键。字符串文字由 maxwell-json.map-null-key.literal 选项定义。                                                                  |
| maxwell-json.map-null-key.literal      | 否    | 'null' | String  | 当 'maxwell-json.map-null-key.mode' 为 LITERAL 时，指定字符串文字以替换空键。                                                                                                                                                                            |
   
#### 支持的Connector
- Kafka
 
#### 示例
使用kafka发送数据，输出到print中。
1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列--\>找到作业的所属队列--\>更多--\>测试地址连通性--\>输入kafka的地址--\>测试）。若能连通，则表示跨源已经绑定成功；否则表示未成功。
2. 创建flink opensource sql作业，选择flink1.12，并提交运行，其代码如下： 
   ```
   create table kafkaSource(
     id bigint,
     name string,
     description string,
     weight DECIMAL(10, 2)  
     ) with (
       'connector' = 'kafka',
       'topic' = '<yourTopic>',
       'properties.group.id' = '<yourGroupId>',
       'properties.bootstrap.servers' = '<yourKafkaAddress1>:<yourKafkaPort>,<yourKafkaAddress2>:<yourKafkaPort>',
       'scan.startup.mode' = 'latest-offset',
       'format' = 'maxwell-json'
   );
   create table printSink(
     id bigint,
     name string,
     description string,
     weight DECIMAL(10, 2)
      ) with (
        'connector' = 'print'
      );
   insert into printSink select * from kafkaSource;
   ```
   
   
3. 向kafka的相应topic中插入下列数据： 
   ```
   {
      "database":"test",
      "table":"e",
      "type":"insert",
      "ts":1477053217,
      "xid":23396,
      "commit":true,
      "position":"master.000006:800911",
      "server_id":23042,
      "thread_id":108,
      "primary_key": [1, "2016-10-21 05:33:37.523000"],
      "primary_key_columns": ["id", "c"],
      "data":{
        "id":111,
        "name":"scooter",
        "description":"Big 2-wheel scooter",
        "weight":5.15
      },
      "old":{
        "weight":5.18
      }
   }
   ```
   
   
4. 用户可按下述操作查看输出结果: 
   - 方法一："更多" -\> "FlinkUI" -\> "Task Managers" -\> "Stdout"。
   
   - 方法二：若在提交运行作业前选择了保存日志，则可以从日志的taskmanager.out文件中查看。
   
   
   ```
   +I(111,scooter,Big 2-wheel scooter,5.15)
   ```