Flink构造宽表实时入库案例介绍

Flink 1.15.4 安装包

Flink cdc的mysql连接器

Flink sql的sdb连接器

MySQL驱动

SDB驱动

Flink jdbc的mysql连接器

tar -zxvf ?flink-1.14.5-bin-scala_2.11.tgz ?-C /opt/

cp sdb-flink-connector-3.4.8-jar-with-dependencies.jar /opt/flink-1.14.5/lib
cp sequoiadb-driver-3.4.8.jre8.jar?/opt/flink-1.14.5/lib
cp flink-sql-connector-mysql-cdc-2.2.1.jar /opt/flink-1.14.5/lib
cp flink-connector-jdbc_2.11-1.14.6.jar /opt/flink-1.14.5/lib

vi conf/flink-conf.yaml

### 配置Master的机器名（IP地址）
jobmanager.rpc.address: sdb1
### 配置每个taskmanager 生成的临时文件夹
io.tmp.dirs: /opt/flink-1.14.5/tmp

vi conf/masters

#作为master的ip和端口号
upgrade1:8081

vi conf/workers

#集群主机名
upgrade1
upgrade2
upgrade3

scp -r /opt/flink-1.14.5 sdbadmin@upgrade2:/opt/
scp -r /opt/flink-1.14.5 sdbadmin@upgrade3:/opt/

[sdbadmin@upgrade1 flink-1.14.5]$ ./bin/start-cluster.sh

[sdbadmin@upgrade1 flink-1.14.5]$ ./bin/sql-client.sh

[sdbadmin@upgrade1 mysql]$ mkdir /opt/sequoiasql/mysql/database/3306/binlog

vim /opt/sequoiasql/mysql/database/3306/auto.cnf

>>配置以下内容：
log_bin=/opt/sequoiasql/mysql/database/3306/binlog
binlog_format=ROW
expire_logs_days=1
server_id=1

[sdbadmin@upgrade1 mysql]$ ./bin/sdb_mysql_ctl stop myinst
[sdbadmin@upgrade1 mysql]$ ./bin/sdb_mysql_ctl start myinst

create database sbtest；
use sbtest；

CREATE TABLE sbtest1 (
????id INT UNSIGNED AUTO_INCREMENT,
????uuid INT(10),
????name1 CHAR(120),
????age INT(4),
????time1 DATETIME,
????PRIMARY KEY(id)
);

CREATE TABLE sbtest2 (
????id INT UNSIGNED AUTO_INCREMENT,
????uuid INT(10),
????name2 CHAR(120),
????age INT(4),
????time1 DATETIME,
????PRIMARY KEY(id)
);

CREATE TABLE sbtest3 (
????id INT UNSIGNED AUTO_INCREMENT,
????uuid INT(10),
????name3 CHAR(120),
????age INT(4),
????time1 DATETIME,
????PRIMARY KEY(id)
);

CREATE TABLE sbtest4 (
????id INT UNSIGNED AUTO_INCREMENT,
????uuid INT(10),
????name1 CHAR(120),
????name2 CHAR(120),
????name3 CHAR(120),
????age INT(4),
????time1 DATETIME,
????PRIMARY KEY(id)
);

CREATE TABLE sbtest1_mysql (
????id INT,
????uuid INT,
????name1 CHAR(120),
????age INT,
????time1 TIMESTAMP,
????PRIMARY KEY (id) NOT ENFORCED
??) WITH (
????'connector' = 'mysql-cdc',
????'hostname' = '192.168.223.135',
????'port' = '3306',
????'username' = 'root',
????'password' = 'root',
????'database-name' = 'sbtest',
????'table-name' = 'sbtest1'
);

CREATE TABLE sbtest2_mysql (
????id INT,
????uuid INT,
????name2 CHAR(120),
????age INT,
????time1 TIMESTAMP,
????PRIMARY KEY (id) NOT ENFORCED
??) WITH (
????'connector' = 'mysql-cdc',
????'hostname' = '192.168.223.135',
????'port' = '3306',
????'username' = 'root',
????'password' = 'root',
????'database-name' = 'sbtest',
????'table-name' = 'sbtest2'
);

CREATE TABLE sbtest3_mysql (
????id INT,
????uuid INT,
????name3 CHAR(120),
????age INT,
????time1 TIMESTAMP,
????PRIMARY KEY (id) NOT ENFORCED
??) WITH (
????'connector' = 'mysql-cdc',
????'hostname' = '192.168.223.135',
????'port' = '3306',
????'username' = 'root',
????'password' = 'root',
????'database-name' = 'sbtest',
????'table-name' = 'sbtest3'
);

CREATE TABLE sbtest4_mysql (
????id BIGINT,
????uuid INT,
????name1 CHAR(120),
????name2 CHAR(120),
????name3 CHAR(120),
????age INT,
????time1 TIMESTAMP,
????PRIMARY KEY (id) NOT ENFORCED
??) WITH (
????'connector' = 'jdbc',
????'url' = 'jdbc:mysql://192.168.223.135:3306/sbtest',
????'username' = 'root',
????'password' = 'root',
????'table-name' = 'sbtest4'
);

CREATE TABLE sbtest_sdb (
????id BIGINT,
????uuid INT,
????name1 CHAR(120),
????name2 CHAR(120),
????name3 CHAR(120),
????age INT,
????time1 TIMESTAMP,
????PRIMARY KEY (id) NOT ENFORCED
) WITH (
????'connector' = 'sequoiadb',
????'bulksize' = '1',
????'hosts' = '192.168.223.135:11810',
????'collectionspace' = 'sbtest',
????'collection' = 'sbtest4'
);

select sdb1.id, sdb1.uuid, sdb1.name1, sdb2.name2, sdb3.name3, sdb1.age, sdb1.time1
from sbtest1_mysql sdb1
left join sbtest2_mysql sdb2
on sdb1.id = sdb2.id
left join sbtest3_mysql sdb3
on sdb1.id = sdb3.id;

insert into sbtest4_mysql select sdb1.id, sdb1.uuid, sdb1.name1, sdb2.name2, sdb3.name3, sdb1.age, sdb1.time1
from sbtest1_mysql sdb1
left join sbtest2_mysql sdb2
on sdb1.id = sdb2.id
left join sbtest3_mysql sdb3
on sdb1.id = sdb3.id;

select sdb1.id, sdb1.uuid, sdb1.name1, sdb2.name2, sdb3.name3, sdb1.age, sdb1.time1
from sbtest1_mysql sdb1
left join sbtest2_mysql sdb2
on sdb1.id = sdb2.id
left join sbtest3_mysql sdb3
on sdb1.id = sdb3.id;

insert into sbtest_sdb select sdb1.id, sdb1.uuid, sdb1.name1, sdb2.name2, sdb3.name3, sdb1.age, sdb1.time1
from sbtest1_mysql sdb1
left join sbtest2_mysql sdb2
on sdb1.id = sdb2.id
left join sbtest3_mysql sdb3
on sdb1.id = sdb3.id;