Flink实时同步MySQL与Doris数据_flink cdc mysql doris

参考：

技术解析｜Doris Connector 结合 Flink CDC 实现 MySQL 分库分表 Exactly Once 精准接入-阿里云开发者社区

逻辑图：

1. Flink环境：

https://flink.apache.org/zh/

• 下载flink-1.15.1

wget https://dlcdn.apache.org/flink/flink-1.15.1/flink-1.15.1-bin-scala_2.12.tgz

• 解压，修改配置

tar -zxvf flink-1.15.1-bin-scala_2.12.tgz cd flink-1.15.1

• 修改配置

修改rest.bind-address为 0.0.0.0

vi conf/flink-conf.yaml

• 下载依赖jar包 至 flink安装目录lib下

wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/flink-sql-connector-mysql-cdc-2.2.1.jar 
 
wget https://repo1.maven.org/maven2/org/apache/doris/flink-doris-connector-1.14_2.12/1.0.3/flink-doris-connector-1.14_2.12-1.0.3.jar

• 启动flink

./bin/start-cluster.sh

• 访问WebUI

http://192.168.0.158:8081

2、MySQL数据表及数据

1. 开启Binlog，进入容器修改/etc/mysql/mysql.cnf，然后重启mysql

[mysqld] 
log_bin=mysql_bin 
binlog-format=Row 
server-id=1

1. 进入MySQL命令行：创建数据库emp，数据表employee：

CREATE DATABASE emp; 
 
USE emp; 
 
CREATE TABLE employee ( 
emp_no INT NOT NULL, 
birth_date DATE NOT NULL, 
first_name VARCHAR(14) NOT NULL, 
last_name VARCHAR(16) NOT NULL, 
gender ENUM ('M','F') NOT NULL, 
hire_date DATE NOT NULL, PRIMARY KEY (emp_no) 
); ​ 
 
INSERT INTO `employee` VALUES 
(10001,'1953-09-02','Georgi','Facello','M','1986-06-26'), 
(10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'), 
(10003,'1959-12-03','Parto','Bamford','M','1986-08-28'), 
(10004,'1954-05-01','Chirstian','Koblick','M','1986-12-01'), 
(10005,'1955-01-21','Kyoichi','Maliniak','M','1989-09-12'), 
(10006,'1953-04-20','Anneke','Preusig','F','1989-06-02'), 
(10007,'1957-05-23','Tzvetan','Zielinski','F','1989-02-10'), 
(10008,'1958-02-19','Saniya','Kalloufi','M','1994-09-15'), 
(10009,'1952-04-19','Sumant','Peac','F','1985-02-18'), 
(10010,'1963-06-01','Duangkaew','Piveteau','F','1989-08-24'), 
(10011,'1953-11-07','Mary','Sluis','F','1990-01-22'), 
(10012,'1960-10-04','Patricio','Bridgland','M','1992-12-18'), 
(10013,'1963-06-07','Eberhardt','Terkki','M','1985-10-20'), 
(10014,'1956-02-12','Berni','Genin','M','1987-03-11'), 
(10015,'1959-08-19','Guoxiang','Nooteboom','M','1987-07-02'), 
(10016,'1961-05-02','Kazuhito','Cappelletti','M','1995-01-27'), 
(10017,'1958-07-06','Cristinel','Bouloucos','F','1993-08-03'), 
(10018,'1954-06-19','Kazuhide','Peha','F','1987-04-03'), 
(10019,'1953-01-23','Lillian','Haddadi','M','1999-04-30'), 
(10020,'1952-12-24','Mayuko','Warwick','M','1991-01-26');

3. Doris数据表

1. 进入MySQL命令行：创建Doris数据库demo，数据表employee_info

CREATE DATABASE demo; 
 
USE demo; 
 
CREATE TABLE employee_info ( 
emp_no int NOT NULL, 
birth_date date, 
first_name varchar(20), 
last_name varchar(20), 
gender char(2), 
hire_date date, 
database_name varchar(50), 
table_name varchar(200) 
) 
UNIQUE KEY(`emp_no`, `birth_date`) 
DISTRIBUTED BY HASH(`birth_date`) BUCKETS 1 
PROPERTIES ( "replication_allocation" = "tag.location.default: 1" );

4. Flink数据表及数据

• 启动fink-sql-client

./bin/sql-client.sh embedded

• 开启Checkpoint

Flink作业周期性执行checkpoint，记录Binlog位点，当作业发生Failover时，便会从之前记录的Binlog位点继续处理。
生产环境建议设置为60秒。

Flink SQL> SET execution.checkpointing.interval = 10s

• 创建MySQL CDC表

Flink SQL> CREATE TABLE employee_source ( 
database_name STRING METADATA VIRTUAL, 
table_name STRING METADATA VIRTUAL, 
emp_no int NOT NULL, 
birth_date date, 
first_name STRING, 
last_name STRING, 
gender STRING, 
hire_date date, 
PRIMARY KEY (`emp_no`) NOT ENFORCED 
) 
WITH ( 
'connector' = 'mysql-cdc', 
'hostname' = 'localhost', 
'port' = '3336', 
'username' = 'root', 
'password' = '1234.abcd', 
'database-name' = 'emp', 
'table-name' = 'employee' 
);

查询数据：

Flink SQL> select * from employee_source limit 10;

• 创建Doris Sink表

Flink SQL> CREATE TABLE cdc_doris_sink ( 
emp_no int , 
birth_date STRING, 
first_name STRING, 
last_name STRING, 
gender STRING, 
hire_date STRING, 
database_name STRING, 
table_name STRING 
) 
WITH ( 
'connector' = 'doris', 
'fenodes' = 'localhost:8030', 
'table.identifier' = 'demo.employee_info', 
'username' = 'root', 
'password' = '1234.abcd' 
);

参数说明：
connector ： 指定连接器是doris
fenodes：doris FE节点IP地址及http port
table.identifier ： Doris对应的数据库及表名
username：doris用户名
password：doris用户密码

查询数据：

Flink SQL> select * from cdc_doris_sink;

• 添加数据同步任务

Flink SQL> insert into cdc_doris_sink (emp_no,birth_date,first_name,last_name,gender,hire_date,database_name,table_name) 
select emp_no,cast(birth_date as string) as birth_date ,first_name,last_name,gender,cast(hire_date as string) as hire_date ,database_name,table_name from employee_source;

WebUI可以看到正在执行中的任务，说明添加完成

查看Doris数据表中数据

mysql> select * from employee_info;

5. 问题说明：

NoResourceAvailableException: Could not acquire the minimum required resources

进入flink目录，修改conf/conf/flink-conf.yaml：taskmanager.numberOfTaskSlots: 4 , 一般配置为cpu的个数。