实时计算Flink版：MySQL整库同步Kafka

使用限制

• 同步的MySQL表必须包含主键。

• 支持使用自建Kafka集群，EMR的Kafka集群和云消息队列 Kafka 版。使用云消息队列 Kafka 版时，只能通过默认接入点使用。

• upsert-kafka表暂未支持作为CTAS和CDAS语法的源表，upsert-kafka表只能作为CTAS和CDAS同步的结果表。

• Kafka集群的存储空间必须大于源表数据的存储空间，否则会因存储空间不足导致数据丢失。因为整库同步Kafka建立的topic都是compacted topic，即topic的每个消息键（Key）仅保留最近的一条消息，但是数据不会过期，compacted topic里相当于保存了与源库的表相同大小的数据。

操作步骤

1. 注册MySQL Catalog和Kafka JSON Catalog。

   详情请参见管理MySQL Catalog和管理Kafka JSON Catalog。

2. 创建并启动一个CDAS或CTAS同步任务，将数据库中的表同步到Kafka中。

   • CDAS同步语句

     CREATE DATABASE IF NOT EXISTS `kafka-catalog`.`kafka`
     AS DATABASE `mysql-catalog`.`database` INCLUDING ALL TABLES;

     说明

     由于Kafka本身没有数据库的概念，所以不存在创建数据库的操作，使用时需要结合IF NOT EXISTS来跳过建库。

   • CTAS同步语句

     CREATE TABLE `kafka-catalog`.`kafka`.`topic`
     AS TABLE `mysql-catalog`.`db`.`table`;

3. 使用同步到Kafka的表。

   整库同步任务建立的Kafka topic名称和MySQL表名相同，分区数和副本数会使用集群的默认配置，并且cleanup.policy会设置为compact。

   使用Kafka JSON Catalog访问MySQL数据库表的对应表，下游作业可以消费Topic中的数据来获取数据库表的最新数据。对于同步到Kafka的表，使用方式有以下两种：

   • 通过Catalog直接使用

     详情请参见使用Kafka JSON Catalog。

     说明

     在直接使用时，由于可能发生了Schema变更，Kafka JSON Catalog解析出的Schema可能与MySQL对应表存在差异，例如出现已经删除的字段，部分字段可能出现为null的情况。

     Catalog读取出的Schema由消费到的数据的字段组成。如果存在删除的字段且消息未过期，则会出现一些已经不存在的字段，这样的字段值会为null，该情况无需特殊处理。

   • 通过创建临时表的方式使用

     这种方式支持用户自定义指定Schema。您可以在Schemas tab页查看并复制表配置的WITH部分。基本格式如下所示。

     CREATE TEMPORARY TABLE tempOrder (
       `key_order_id` BIGINT NOT NULL,
       `value_product` STRING,
       PRIMARY KEY (key_order_id) NOT ENFORCED
     ) WITH (
       'connector' = 'upsert-kafka',
       'topic' = 'order',
       'properties.bootstrap.servers' = 'xxxx',
       'key.format' = 'json',
       'key.fields-prefix' = 'key_',
       'value.format' = 'json',
       'value.fields-prefix' = 'value_',
       'value.fields-include' = 'EXCEPT_KEY',
       'value.json.infer-schema.flatten-nested-columns.enable' = 'false',
       'value.json.infer-schema.primitive-as-string' = 'false'
     );

     参数	说明	备注
     connector	Connector类型。	固定值为upsert-kafka。
     topic	对应的Topic名称。	和Kafka JSON Catalog的描述保持一致。
     properties.bootstrap.servers	Kafka Broker地址。	格式为host:port,host:port,host:port，以英文逗号（,）分割。
     key.format	Flink Kafka Connector在序列化或反序列化Kafka的消息键（Key）时使用的格式。	固定值为json。
     key.fields-prefix	为所有Kafka消息键（Key）指定自定义前缀，以避免与消息体（Value）或Metadata字段重名。	需要和Kafka JSON Catalog的key.fields-prefix参数值保持一致。
     value.format	Flink Kafka Connector在序列化或反序列化Kafka的消息体（Value）时使用的格式。	固定值为json。
     value.fields-prefix	为所有Kafka消息体（Value）指定自定义前缀，以避免与消息键（Key）或Metadata字段重名。	需要和Kafka JSON Catalog的value.fields-prefix参数值保持一致。
     value.fields-include	定义消息体在处理消息键字段时的策略。	固定值为EXCEPT_KEY。表示消息体中不包含消息键的字段。
     value.json.infer-schema.flatten-nested-columns.enable	Kafka消息体（Value）是否递归式地展开JSON中的嵌套列。	对应Catalog的infer-schema.flatten-nested-columns.enable参数配置值。
     value.json.infer-schema.primitive-as-string	Kafka消息体（Value）是否推导所有基本类型为String类型。	对应Catalog的infer-schema.primitive-as-string参数配置值。

应用示例

-- 将订单信息和用户表做join，展示每个订单的用户名和商品名。
SELECT order.id as order_id, product, user.name as user_name
FROM order LEFT JOIN user
ON order.user_id = user.id;

-- 将评论和用户表做join，展示每个评论的内容和对应用户名。
SELECT feedback.id as feedback_id, comment, user.name as user_name
FROM feedback LEFT JOIN user
ON feedback.user_id = user.id;

对于以上两个SQL任务，user表在两个作业中都被使用了一次。运行时，两个作业都会读取MySQL的全量数据和增量数据。全量读取需要创建MySQL连接，增量读取需要创建Binlog Client。随着作业的不断增多，MySQL连接和Binlog Client资源也会对应增长，会给上游数据库产生极大的压力。

CREATE DATABASE IF NOT EXISTS `kafka-catalog`.`kafka`
AS DATABASE `mysql-catalog`.`database` INCLUDING ALL TABLES;

-- 将订单信息和Kafka JSON Catalog中的用户表做join，展示每个订单的用户名和商品名。
SELECT order.id as order_id, product, user.value_name as user_name
FROM order LEFT JOIN `kafka-catalog`.`kafka`.`user` as user
ON order.user_id = user.id;

-- 将评论和Kafka JSON Catalog中的用户表做join，展示每个评论的内容和对应用户名。
SELECT feedback.id as feedback_id, comment, user.value_name as user_name
FROM feedback LEFT JOIN `kafka-catalog`.`kafka`.`user` as user
ON feedback.user_id = user.id;

相关文档

• CREATE TABLE AS（CTAS）语句

• CREATE DATABASE AS（CDAS）语句

• MySQL的CDC源表