Flink实时同步MySQL与Doris数据

原创
08/11 11:36
阅读数 544

逻辑图:

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

2. 进入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的个数。

 

** 自荐 **

我的 Gitee,欢迎来虐

展开阅读全文
加载中

作者的其它热门文章

打赏
0
0 收藏
分享
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部