1. 創(chuàng)建MySQL庫表，寫入demo數(shù)據(jù)
- 登錄測試MySQL
mysql -u root -pnew_password
- 創(chuàng)建MySQL庫表，寫入demo數(shù)據(jù)
CREATE DATABASE emp_1;
USE emp_1;
CREATE TABLE employees_1 (
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 `employees_1` VALUES (10001,'1953-09-02','Georgi','Facello','M','1986-06-26'),
(10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'),
(10036,'1959-08-10','Adamantios','Portugali','M','1992-01-03');
注意：MySQL需要開通bin-log
- log_bin=mysql_bin
- binlog-format=Row
- server-id=1
2. 創(chuàng)建Doris庫表
- 創(chuàng)建Doris表
mysql -uroot -P9030 -h127.0.0.1
create database demo;
use demo;
CREATE TABLE all_employees_info (
emp_no int NOT NULL,
birth_date date,
first_name varchar(20),
last_name varchar(20),
gender char(2),
hire_date date
)
UNIQUE KEY(`emp_no`, `birth_date`)
DISTRIBUTED BY HASH(`birth_date`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);
3. 啟動Flink
- 啟動flink
cd /mnt/apps/flink-1.15.3/
#啟動flink，這里服務(wù)已經(jīng)啟動
bin/start-cluster.sh
#進(jìn)入SQL控制臺
bin/sql-client.sh embedded
- 創(chuàng)建Flink 任務(wù)：
SET 'execution.checkpointing.interval' = '10s';
CREATE TABLE employees_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' = '3306',
'username' = 'root',
'password' = 'new_password',
'database-name' = 'emp_1',
'table-name' = 'employees_1'
);
CREATE TABLE cdc_doris_sink (
emp_no int ,
birth_date STRING,
first_name STRING,
last_name STRING,
gender STRING,
hire_date STRING
)
WITH (
'connector' = 'doris',
'fenodes' = '172.16.64.9:8030',
'table.identifier' = 'demo.all_employees_info',
'username' = 'root',
'password' = '',
'sink.properties.two_phase_commit'='true',
'sink.label-prefix'='doris_demo_emp_002'
);
insert into cdc_doris_sink (emp_no,birth_date,first_name,last_name,gender,hire_date)
select emp_no,cast(birth_date as string) as birth_date ,first_name,last_name,gender,cast(hire_date as string) as hire_date from employees_source;
-
輸入如下地址，查看flink任務(wù)
http://localhost:8081/#/job/running -
數(shù)據(jù)驗(yàn)證：啟動后可以看到有數(shù)據(jù)實(shí)時進(jìn)入Doris了文章來源：http://www.zghlxwxcb.cn/news/detail-409930.html
mysql -uroot -P9030 -h127.0.0.1
mysql> select * from all_employees_info;
+--------+------------+------------+-----------+--------+------------+
| emp_no | birth_date | first_name | last_name | gender | hire_date |
+--------+------------+------------+-----------+--------+------------+
| 10001 | 1953-09-02 | Georgi | Facello | M | 1986-06-26 |
| 10002 | 1964-06-02 | Bezalel | Simmel | F | 1985-11-21 |
| 10036 | 1959-08-10 | Adamantios | Portugali | M | 1992-01-03 |
| 20001 | 1953-09-02 | Georgi | Facello | M | 1986-06-26 |
+--------+------------+------------+-----------+--------+------------+
4 rows in set (0.02 sec)
Link
- https://zhuanlan.zhihu.com/p/532913664
- https://www.runoob.com/mysql/mysql-install.html
- https://repo.maven.apache.org/maven2/org/apache/doris/flink-doris-connector-1.15/1.2.1/
- https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/
Jar包地址：
flink 環(huán)境：1.15.3文章來源地址http://www.zghlxwxcb.cn/news/detail-409930.html
- https://dlcdn.apache.org/flink/flink-1.15.3/flink-1.15.3-bin-scala_2.12.tgz
解壓并將jar包防止在Flink 的lib下
flink-doris-connector：1.15 - https://repo.maven.apache.org/maven2/org/apache/doris/flink-doris-connector-1.15/1.2.1/flink-doris-connector-1.15-1.2.1.jar
cdc mysql：flink-sql-connector-mysql-cdc-2.2.1.jar - https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/flink-sql-connector-mysql-cdc-2.2.1.jar
到了這里，關(guān)于Doris通過Flink CDC接入MySQL實(shí)戰(zhàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
