下载flink 1.20

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

解压flink 1.20

tar -xzvf flink-1.20.3-bin-scala_2.12.tgz

配置flink

[dmdba@localhost ~]$ vi /home/dmdba/flink-1.20.3/conf/config.yaml

将localhost全部替换为0.0.0.0

%s/localhost/0.0.0.0/g

配置数据库驱动包

默认已经存在运行的数据库

下载以下两个包，存放到flink的lib目录
flink-connector-jdbc-dameng-3.3.0-1.20.jar（必须）
flink-connector-jdbc-3.3.0-1.20.jar（如果代码中存在，可以不用下载）

找到数据库版本对应的jdbc.jar包存放到flink/lib目录

lib目录完整截图如下

注意下目录和文件的权限。

启动flink

cd /home/dmdba/flink-1.20.3
./bin/start-cluster.sh

登录数据库创建相应表

create table dm_test(c1 char(255) primary key);
create table my_table(id INT,name VARCHAR2(100),age INT);

启动sql-client

cd /home/dmdba/flink-1.20.3
./bin/sql-client.sh

在sql-client界面执行以下命令

create table dm_test(C1 STRING) WITH ('connector' = 'dm-jdbc','url' =    'jdbc:dm://ip地址:端口/模式','table-name' = '表名', 'driver' = 'dm.jdbc.driver.DmDriver','username' = '用户名', 'password'= '密码');
CREATE TABLE my_table(id INT,name STRING,age INT) WITH ('connector' = 'jdbc','url' =    'jdbc:dm://ip地址:端口/模式','table-name' = '表名', 'driver' = 'dm.jdbc.driver.DmDriver','username' = '用户名', 'password'= '密码');

截图如下：

flink 1.70的驱动包如下：

flink-connector-dameng-3.1.1-1.17.jar
flink-connector-jdbc-3.1.jar