flink部署使用(flink-connector-jdbc)连接达梦数据库并写入读取数据

flink介绍

1）Apache Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink 被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。

2）在实时计算或离线任务中，往往需要与关系型数据库交互，例如 MySQL、PostgreSQL 等。Apache Flink 提供了 JDBC Connector，可以方便地将流式数据写入或读取数据库。

3）flink版本下载：https://archive.apache.org/dist/flink/

flink单机搭建

## 1. 下载并解压flink
[root@localhost flink_soft]# mkdir /data/flink_soft
[root@localhost flink_soft]# tar -zxvf flink-1.16.1-bin-scala_2.12.tgz
## 2. 修改配置文件，把下面的三行全部去掉
[root@localhost flink-1.16.1]# cd /data/flink_soft/flink-1.16.1
[root@localhost flink-1.16.1]# vim /data/flink_soft/flink-1.16.1/conf/flink-conf.yaml
rest.port: 8081
rest.address: 0.0.0.0
rest.bind-address: 0.0.0.0
## 3. 启动flink
[root@localhost flink-1.16.1]# ./bin/start-cluster.sh
## 4. 查询进程是否存在
[root@localhost flink-1.16.1]# ps aux | grep flink
## 5. 访问http://192.168.112.162:8081/ 即可。

将已经适配dameng的jar包放到lib目录下

1）下载已经适配好的包https://github.com/gaoyuan98/flink-connector-jdbc-dameng/releases

提供了两个版本的dameng适配驱动包，一个是实现JdbcFactory接口，还有一个是实现JdbcDialectFactory接口。

2）截止发文v3.3版本官方还未正式发版，所以大概率是用这个版本：flink-connector-jdbc-dameng_20250331_(适用于v3.2及以下版本)

3）将下载好的适配包放到flink的lib目录下

DmJdbcDriver8.jar 达梦数据库jdbc驱动，可以更换为与数据库版本相同的驱动。

flink-connector-jdbc-3.1.jar flink使用jdbc方式连接数据库时的桥接包，如果项目本身已经有flink-connector-jdbc包可忽略该包。

flink-connector-jdbc-dameng-1.0.jar flink使用jdbc方式连接达梦数据库的适配包，源码基于flink-connector-jdbc.jar包进行调整，所以该包必须存在。

如项目中已经有flink-connector-jdbc的包，那么只需要使用DmJdbcDriver8.jar跟flink-connector-jdbc-dameng-1.0.jar的驱动包即可。

如项目中没有flink-connector-jdbc的包，就把这三个包全部放到lib下。

[root@localhost lib]# cd /data/flink_soft/flink-1.16.1/lib
[root@localhost lib]# ll
total 204020
-rw-r--r--. 1 root root   1615303 Jan 17 00:30 DmJdbcDriver8.jar
-rwxrwxrwx. 1 root root    198857 Jan 19  2023 flink-cep-1.16.1.jar
-rwxrwxrwx. 1 root root    516144 Jan 19  2023 flink-connector-files-1.16.1.jar
-rw-r--r--. 1 root root    277945 Mar 28 23:46 flink-connector-jdbc-3.1-SNAPSHOT.jar
-rw-r--r--. 1 root root     13458 Mar 29 00:13 flink-connector-jdbc-dameng-1.0-SNAPSHOT.jar
-rwxrwxrwx. 1 root root    102470 Jan 19  2023 flink-csv-1.16.1.jar
-rwxrwxrwx. 1 root root 117107159 Jan 19  2023 flink-dist-1.16.1.jar
-rwxrwxrwx. 1 root root    180248 Jan 19  2023 flink-json-1.16.1.jar
-rwxrwxrwx. 1 root root  21052640 Jan 19  2023 flink-scala_2.12-1.16.1.jar
-rwxrwxrwx. 1 root root  10737871 Jan 13  2023 flink-shaded-zookeeper-3.5.9.jar
-rwxrwxrwx. 1 root root  15367504 Jan 19  2023 flink-table-api-java-uber-1.16.1.jar
-rwxrwxrwx. 1 root root  36249667 Jan 19  2023 flink-table-planner-loader-1.16.1.jar
-rwxrwxrwx. 1 root root   3133690 Jan 19  2023 flink-table-runtime-1.16.1.jar
-rwxrwxrwx. 1 root root    208006 Jan 13  2023 log4j-1.2-api-2.17.1.jar
-rwxrwxrwx. 1 root root    301872 Jan 13  2023 log4j-api-2.17.1.jar
-rwxrwxrwx. 1 root root   1790452 Jan 13  2023 log4j-core-2.17.1.jar
-rwxrwxrwx. 1 root root     24279 Jan 13  2023 log4j-slf4j-impl-2.17.1.jar

重启flink

[root@localhost flink-1.16.1]# cd /data/flink_soft/flink-1.16.1
[root@localhost flink-1.16.1]# ./bin/stop-cluster.sh
[root@localhost flink-1.16.1]# ./bin/start-cluster.sh## 如果报错的话查看这个日志
tail -f $FLINK_HOME/log/flink-*-taskexecutor-*.log

flink驱动验证

在达梦数据库上创建表数据

CREATE TABLE source_table (id INT PRIMARY KEY,name VARCHAR(50),age INT
);
INSERT INTO source_table (id, name, age) VALUES (1, 'Alice', 30);
INSERT INTO source_table (id, name, age) VALUES (2, 'Bob', 25);
INSERT INTO source_table (id, name, age) VALUES (3, 'Charlie', 40);
COMMIT;

在 Flink SQL CLI 中定义达梦表

[root@localhost lib]# cd /data/flink_soft/flink-1.16.1/
[root@localhost flink-1.16.1]#  ./bin/sql-client.sh embeddedCREATE TABLE source (id INT,name STRING,age INT
) WITH ('connector' = 'jdbc','url' = 'jdbc:dm://192.168.127.2:5236/SYSDBA','table-name' = 'source_table','driver' = 'dm.jdbc.driver.DmDriver','username' = 'SYSDBA','password' = 'SYSDBA123'
);## 在 Flink SQL CLI 中查询数据
SELECT * FROM source;
## 筛选数据，比如 查询年龄大于 30 的用户：
SELECT id, name FROM source WHERE age > 30;
## 插入数据
INSERT INTO source (id, name, age) VALUES (3, '33', 33);

CREATE TABLE source1 (id INT,name STRING,age INT
) WITH ('connector' = 'dameng','url' = 'jdbc:dm://81.70.105.201:5236/SYSDBA','table-name' = 'source_table','driver' = 'dm.jdbc.driver.DmDriver','username' = 'SYSDBA','password' = '123456'
);
SELECT * FROM source1;