flink cdc demo

Flink 1.11 发布了支持 MySQL CDC（Changelog Data Capture） 可以直接解析 Canal / Debezium 格式的 MySQL Binlog

对于实时处理程序，MySQL 的 Binlog 是很重要的数据源（上一个项目几乎所有实时数据都是来自业务系统的数据库，
也就是MySQL，算上分库分表，接了上千个 MySQL 表的 binlog）

Flink 1.11 的 CDC 发布之后，第一时间就尝试了一下 Canal 的 binlog 格式，不过感觉一般，还要部署解析的 Canal / Debezium，我们自己开发的解析 MySQL binlog 的组件，
比 Cancel 适合更适合我们，就放弃继续研究这个东西了

最近一段时间，也看到很多同学在社区提起 Flink CDC ，一直以为是类似于 Stateful Functions 的 Flink 组件，今天看下下，才发现是云邪大佬开源的 Flink-cdc-connector

GitHub 地址： https://github.com/ververica/flink-cdc-connectors

Flink-cdc-connector 就是个 CDC 组件，可以跳过 Canal / Debezium 等解析 Binlog 的工具，直接获取 MySQL 的 Binlog，转换成 Flink 的流表

Flink-cdc-connector 目前支持 MySQL 和 Postgres 两种数据库，详情查看 GitHub

----------------进入主题--------------------------

这次是 cdc 的demo,所以需要将数据写到 MySQL 中, 所以我就直接起了两个任务:

1、 kafka to mysql
2、 mysql cdc to kafka

## kafka to mysql

数据还是使用的之前从天池公开数据集中下载的 user_log，kafka source 没什么好说的，直接从 kafka 读取数据，使用 jdbc sink 写入到 mysql 中

mysql 表结构：

create table user_log
(
    id          int auto_increment primary key,
    user_id     varchar(20) not null,
    item_id     varchar(20) null,
    category_id varchar(20) null,
    behavior    varchar(10) null,
    ts          datetime    null
)
comment '天池公开数据集-淘宝用户数据';

flink jdbc sink 表如下：

CREATE TABLE mysql_table_venn_user_log_sink (
  user_id STRING
  ,item_id STRING
  ,category_id STRING
  ,behavior STRING
  ,ts timestamp(3)
) WITH (
  'connector' = 'jdbc'
  ,'url' = 'jdbc:mysql://venn:3306/venn'
  ,'table-name' = 'user_log'
  ,'username' = 'root'
  ,'password' = '123456'
  ,'sink.buffer-flush.max-rows' = '100' -- default
  ,'sink.buffer-flush.interval' = '1s'
  ,'sink.max-retries' = '3'
);

写入数据如下：

 ## mysql cdc to kafka

这个要稍微麻烦一点，由于 cdc source 是 upsert 的，并且不支持 WindowGroupAgg，所以使用了自己开发的 kafka upsert sink（注意： 代码中直接过滤了 delete 流）

The main method caused an error: GroupWindowAggregate doesn't support consuming update and delete changes which is produced by node TableSourceScan(table=[[default_catalog, default_database, cdc_mysql_venn_user_log]], fields=[id, user_id, item_id, category_id, behavior, ts])
        at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:302)

源码： KafkaUpsertTableSink

@Override
public DataStreamSink<?> consumeDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {

    final SinkFunction<Row> kafkaProducer = createKafkaProducer(
            topic,
            properties,
            serializationSchema,
            partitioner);

    // todo cast DataStream<Tuple2<Boolean, Row>> to DataStream<Row>
    return dataStream
            .flatMap(new FlatMapFunction<Tuple2<Boolean, Row>, Row>() {
                @Override
                public void flatMap(Tuple2<Boolean, Row> element, Collector<Row> out) throws Exception {
                    // upsertStream include insert/update/delete change, true is upsert, false is delete
                    // create new row include upsert message
                    if (element.f0) {
                        out.collect(element.f1);
                    } else {
                        System.out.println("KafkaUpsertTableSinkBase : retract stream f0 will be false");
                    }
                }
            })
            .addSink(kafkaProducer)
            .setParallelism(dataStream.getParallelism())
            .name(TableConnectorUtils.generateRuntimeName(this.getClass(), getFieldNames()));
}

下面来看 cdc 的 sql:

-- creates a mysql mysql table source
drop table if exists cdc_mysql_venn_user_log;
CREATE TABLE cdc_mysql_venn_user_log (
  id varchar
  ,user_id VARCHAR
  ,item_id VARCHAR
  ,category_id VARCHAR
  ,behavior VARCHAR
  ,ts TIMESTAMP(3)
  ,proc_time as PROCTIME()
  ,PRIMARY KEY (id) NOT ENFORCED
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = 'venn',
 'port' = '3306',
 'username' = 'root',
 'password' = '123456',
 'database-name' = 'venn',
 'table-name' = 'user_log'
);

-- kafka sink
drop table if exists cdc_mysql_user_log_sink;
CREATE TABLE cdc_mysql_user_log_sink (
  id varchar
  ,user_id VARCHAR
  ,item_id VARCHAR
  ,category_id VARCHAR
  ,behavior VARCHAR
  ,ts TIMESTAMP(3)
) WITH (
  'connector.type' = 'upsertKafka'
  ,'connector.version' = 'universal'
  ,'connector.topic' = 'cdc_mysql_user_log_sink'
  ,'connector.properties.zookeeper.connect' = 'venn:2181'
  ,'connector.properties.bootstrap.servers' = 'venn:9092'
  ,'format.type' = 'json'
);

-- sink to kafka
insert into cdc_mysql_user_log_sink
select id, user_id, item_id, category_id, behavior, ts
from cdc_mysql_venn_user_log;

注： flink cdc connector 表不支持定义 watermark

java.lang.UnsupportedOperationException: Currently, defining WATERMARK on a changelog source is not supported.

删除mysql 表数据时，taskmanager 打印 delete message：

完整代码请查看： GitHub https://github.com/springMoon/sqlSubmit

拓展： Flink SQL CDC 上线！我们总结了 13 条生产实践经验 https://mp.weixin.qq.com/s/Mfn-fFegb5wzI8BIHhNGvQ

欢迎关注Flink菜鸟公众号，会不定期更新Flink（开发技术）相关的推文