Flink实时同步MySQL与Doris数据

缺乏、安全感 2024-03-26 08:08 86阅读 0赞

## 参考： ##

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

## 逻辑图： ##

![92ff3301f98e82db69ce53b58354d864.jpeg][]

**1. Flink环境：**

[https://flink.apache.org/zh/][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

![caa42c0ba7c462caa527a176f9826910.png][]

*  访问WebUI

[http://192.168.0.158:8081][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

![a61809e6f460ab1899801342ad650349.png][]

*  开启Checkpoint

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

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

![4cc8317cf5cfd67470d6b1f0329bf197.png][]

*  创建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;

![format_png][]

*  创建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;

![format_png 1][]

*  添加数据同步任务

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可以看到正在执行中的任务，说明添加完成

![0324aeb24fd97cf06b2d937060038d4f.png][]

查看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的个数。

[Doris Connector _ Flink CDC _ MySQL _ Exactly Once _-]: https://developer.aliyun.com/article/1005812?scm=20140722.ID_community@@article@@1005812.P_121.MO_938-ST_5186-V_1-ID_community@@article@@1005812-OR_rec
[92ff3301f98e82db69ce53b58354d864.jpeg]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/fd2a5802c187479b833caae5f1a772e7.png
[https_flink.apache.org_zh]: https://link.zhihu.com/?target=https%3A//flink.apache.org/zh/
[caa42c0ba7c462caa527a176f9826910.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/7310f572f79a46ef91baab2916aa8b86.png
[http_192.168.0.158_8081]: https://link.zhihu.com/?target=http%3A//192.168.0.158%3A8081/
[a61809e6f460ab1899801342ad650349.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/0896af4f9e95465c8944a4739c63f77d.png
[4cc8317cf5cfd67470d6b1f0329bf197.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/a0637bde77184eb89ed69a393d312699.png
[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/e9bc11fe3ffc455d9654badcfefdbce6.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/f6e94445a9cf4f5eb7c93a1ce0e8b4c8.png
[0324aeb24fd97cf06b2d937060038d4f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/30ca4a13d232466fa3005f830e30b514.png