FlinkCDC 达梦数据库实时同步

一、Flink部署

1.1、JAVA环境

1. vi /etc/profile
2. export JAVA_HOME=/data/flinkcdc/jdk1.8.0_181
3. export CLASSPATH=$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
4. export PATH=$JAVA_HOME/bin:$PATH
6. source /etc/profile
8. vi ~/.bash_profile
10. export FLINK_HOME=/data/flinkcdc/flink-1.17.0
11. export PATH=$PATH:$FLINK_HOME/bin
13. source ~/.bash_profile

复制代码

1.2、设置Flink

1. vim conf/flink-conf.yaml
2. 添加配置：env.java.home=/data/flinkcdc/jdk1.8.0_181
5. ①、localhost  修改为IP地址
6. rest.port: 8088
7. rest.address: 192.168.33.231
8. ②、关闭防火墙
9. systemctl status firewalld
10. systemctl stop firewalld

复制代码

1.3、Flink CDC Jar包

CDC jar放到Flink安装包解压之后的lib目次
1.4、启动flink

1. bin/start-cluster.sh
3. Flink Web-UI
4. http://192.168.33.231:8088

复制代码

1.5、启动 Flink SQL CLI

1. bin/sql-client.sh

复制代码

二、达梦数据库搭建

2.1、docker dm8

1. docker run -d \
2. --name dm8 \
3. --restart=always \
4. --privileged=true \
5. -e LD_LIBRARY_PATH=/opt/dmdbms/bin \
6. -e PAGE_SIZE=16 \
7. -e EXTENT_SIZE=32 \
8. -e LOG_SIZE=1024 \
9. -e CASE_SENSITIVE=0 \
10. -e UNICODE_FLAG=1 \
11. -e INSTANCE_NAME=DM8_CDC \
12. -e SYSDBA_PWD=SYSDBA001 \
13. -v /docker/dm8_data_cdc:/opt/dmdbms/data \
14. -p 5236:5236 \
15. dm8_flinkcdc:dm8

复制代码

查察容器运行情况

1. 查看数据库容器
2. lsof -i:5236
3. docker logs -f  dm8
4. docker exec -it dm8 bash

复制代码

2.2、开启达梦日志归档

1. ##查看当前数据库是否开启归档
2. select arch_mode from v$database;
3. ##查询有哪些归档日志
4. SELECT NAME , FIRST_TIME , NEXT_TIME , FIRST_CHANGE# , NEXT_CHANGE# FROM V$ARCHIVED_LOG;　
5. SELECT * FROM V$ARCH_FILE
8. ##修改数据库实例的 /dmdata/DAMEGN/dm.ini文件中 ARCH_INI 参数值
9. vi /dmdata/DAMENG/dm.ini
10. ##将 ARCH_INI 值改为 1，保存后退出
11. ARCH_INI = 1 #开启归档功能
12. RLOG_APPEND_LOGIC = 1
15. ##新增文件dmarch.ini
16. vi /dmdata/DAMENG/dmarch.ini
17. ##新增如下内容
18. [ARCHIVE_LOCAL1]
19. ARCH_TYPE = LOCAL
20. ARCH_DEST = /dmarch
21. ARCH_FILE_SIZE = 2048
22. ARCH_SPACE_LIMIT = 102400
24. ##最后重启数据库完成归档配置
28. #DaMeng Database Archive Configuration file
29. #this is comments
31.         ARCH_WAIT_APPLY      = 0
33. [ARCHIVE_LOCAL1]
34.         ARCH_TYPE            = LOCAL
35.         ARCH_DEST            = /opt/dmdbms/data/DAMENG/arch
36.         ARCH_FILE_SIZE       = 1024
37.         ARCH_SPACE_LIMIT     = 51200
38.         ARCH_FLUSH_BUF_SIZE  = 0
39.         ARCH_HANG_FLAG       = 1

复制代码

 2.3、重启dm8数据库

1. docker restart dm8

复制代码

三、实时同步测试

1. ##达梦
2. CREATE TABLE t_source_dm (
3.     id INT,
4.     name VARCHAR,
5.     insert_date DATE,
6.     PRIMARY KEY (id) NOT ENFORCED
7. ) WITH (
8.   'connector' = 'dm',
9.   'startupOptions' = 'Initial',
10.   'hostname' = '192.168.33.231',
11.   'port' = '5236',
12.   'username' = 'SYSDBA',
13.   'password' = 'SYSDBA001',
14.   'database' = 'DM8_CDC',
15.   'schema' = 'SYSDBA',
16.   'table' = 'dm_flinkcdc'
17. );
20. ##MYSQL
21. CREATE TABLE sink_mysql_test (
22.   id int,
23.   name STRING,
24.   insert_date date,
25.   PRIMARY KEY (id) NOT ENFORCED
26. ) WITH (
27.    'connector' = 'jdbc',
28.    'url' = 'jdbc:mysql://192.168.33.231:3306/flinkcdc',
29.    'driver' = 'com.mysql.cj.jdbc.Driver',
30.    'username' = 'root',
31.    'password' = 'YTP1101102233',
32.    'table-name' = 'dm_to_mysql'
33. );
35. insert into sink_mysql_test
36. select * from t_source_dm;

复制代码

四、FlinkCDC 达梦数据库 所需文件下载

4.1、所需Jar包

4.2、支持JAVA步伐和SQL

 

4.3、完成步伐和说明文档下载地址

https://download.csdn.net/download/ytp552200ytp/90103896
如果着实没有CSDN积分，配景接洽我留下邮箱，我看到后私信发到邮箱，支持共享。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。