FlinkCDC数据准备

1. 在MySQL中创建数据库及表

create database test;
create database test_route;
CREATE TABLE test.t1(
  `id` VARCHAR(255) PRIMARY KEY,
  `name` VARCHAR(255)
);
CREATE TABLE test.t2(
`id` VARCHAR(255) PRIMARY KEY,
`name` VARCHAR(255)
);
CREATE TABLE test.t3(
`id` VARCHAR(255) PRIMARY KEY,
`sex` VARCHAR(255)
);
CREATE TABLE test_route.t1(
`id` VARCHAR(255) PRIMARY KEY,
`name` VARCHAR(255)
);
CREATE TABLE test_route.t2(
`id` VARCHAR(255) PRIMARY KEY,
`name` VARCHAR(255)
);
CREATE TABLE test_route.t3(
`id` VARCHAR(255) PRIMARY KEY,
`sex` VARCHAR(255)
);

2. 插入数据

在test数据库中插入数据

use test;
INSERT INTO t1 VALUES('1001','zhangsan');
INSERT INTO t1 VALUES('1002','lisi');
INSERT INTO t1 VALUES('1003','wangwu');

INSERT INTO t2 VALUES('1001','zhangsan');
INSERT INTO t2 VALUES('1002','lisi');
INSERT INTO t2 VALUES('1003','wangwu');

INSERT INTO t3 VALUES('1001','F');
INSERT INTO t3 VALUES('1002','F');
INSERT INTO t3 VALUES('1003','M');

在test_route数据库中插入数据

use test_route;
INSERT INTO t1 VALUES('1001','zhangsan');
INSERT INTO t1 VALUES('1002','lisi');
INSERT INTO t1 VALUES('1003','wangwu');

INSERT INTO t2 VALUES('1004','zhangsan');
INSERT INTO t2 VALUES('1005','lisi');
INSERT INTO t2 VALUES('1006','wangwu');

INSERT INTO t3 VALUES('1001','F');
INSERT INTO t3 VALUES('1002','F');
INSERT INTO t3 VALUES('1003','M');

3. 开启MySQL Binlog并重启MySQL

sudo vim /etc/my.cnf
#添加如下配置信息,开启`test`以及`test_route`数据库的Binlog
#数据库id
server-id = 1
##启动binlog，该参数的值会作为binlog的文件名
log-bin=mysql-bin
##binlog类型，maxwell要求为row类型
binlog_format=row
##启用binlog的数据库，需根据实际情况作出修改
binlog-do-db=test
binlog-do-db=test_route