Maxwell Format

功能描述

Maxwell是一个CDC（Changelog Data Capture）工具，可以将MySql中的更改实时流式写入到Kafka等流式connector。Maxwell为changelog提供了统一的格式，而且支持使用JSON对消息进行序列化。

Flink 支持将 Maxwell JSON 消息解释为 INSERT/UPDATE/DELETE 消息到 Flink SQL 系统中。在许多情况下，这对于利用此功能很有用。

支持的Connector

• Kafka
• Filesystem

注意事项

Maxwell应用允许将每个变动消息精确地传递一次。在这种情况下，Flink在消费Maxwell生成的消息时处理得很好。如果Maxwell应用程序在at-least-once模式处理，它可能向Kafka写入重复的改动消息，Flink将获得重复的消息。这可能会导致Flink查询得到错误的结果或意外的异常。因此，在这种情况下，建议将作业配置table.exec.source.cdc-events-duplicate设置为true，并在源表上定义PRIMARY KEY。Framework将生成一个额外的有状态操作符，并使用主键对变更事件进行去重，并生成一个规范化的changelog流。

参数说明

元数据

元数据可以在 DDL 中作为只读（虚拟）meta 列声明。

元数据使用示例如下：

CREATE TABLE KafkaTable (
  origin_database STRING METADATA FROM 'value.database' VIRTUAL,
  origin_table STRING METADATA FROM 'value.table' VIRTUAL,
  origin_primary_key_columns ARRAY<STRING> METADATA FROM 'value.primary-key-columns' VIRTUAL,
  origin_ts TIMESTAMP(3) METADATA FROM 'value.ingestion-timestamp' VIRTUAL,
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'kafkaTopic',
  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
  'properties.group.id' = 'GroupId',
  'scan.startup.mode' = 'earliest-offset',
  'value.format' = 'maxwell-json'
);

示例

使用kafka发送数据，输出到print中。

1. 根据kafka所在的虚拟私有云和子网创建相应的跨源，并绑定所要使用的队列。然后设置安全组，入向规则，使其对当前将要使用的队列放开，并根据kafka的地址测试队列连通性（通用队列 > 找到作业的所属队列 > 更多 > 测试地址连通性 > 输入kafka的地址 > 测试）。如果能连通，则表示跨源已经绑定成功；否则表示未成功。
2. 创建flink opensource sql作业，选择flink1.15，并提交运行，其代码如下：

   CREATE TABLE kafkaSource (
     id bigint,
     name string,
     description string,  
     weight DECIMAL(10, 2)
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'kafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     '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中插入下列数据（每个字段的含义请参考Maxwell documentation）：

   {
      "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. 按照如下方式查看taskmanager.out文件中的数据结果：
   1. 登录DLI管理控制台，选择“作业管理 > Flink作业”。
   2. 单击对应的Flink作业名称，选择“运行日志”，单击“OBS桶”，根据作业运行的日期，找到对应日志的文件夹。
   3. 进入对应日期的文件夹后，找到名字中包含“taskmanager”的文件夹进入，下载获取.out文件查看结果日志。

   +I[111, scooter, Big 2-wheel scooter, 5.15]