FlinkSQL CDC实现同步oracle数据到mysql

环境准备
1、flink 1.13.0
2、oracle 11g
3、flink-connector-oracle-cdc 2.1.0

1、oracle环境配置

首先需要安装oracle环境，参考 https://blog.csdn.net/qq_36039236/article/details/124224500?spm=1001.2014.3001.5502

进入容器进行配置:

docker exec -it oracle11 bash
# 切换到oracle用户
su - oracle
# 创建数据需要的目录,需要提前创建，否则报错目录不存在
mkdir /home/oracle/oracle-data-test
sqlplus /nolog
# 以 DBA 身份连接到数据库
SQL> conn /as sysdba
1
2
3
4
5
6
7
8

接下来进行相关重要的配置:

-- 启用日志归档
alter system set db_recovery_file_dest_size = 10G;
alter system set db_recovery_file_dest = '/home/oracle/oracle-data-test' scope=spfile;
shutdown immediate;
startup mount;
alter database archivelog;
alter database open;

-- 检查日志归档是否开启
archive log list;

-- 为捕获的数据库启用补充日志记录,以便数据更改捕获更改的数据库行之前的状态,下面说明了如何在数据库级别进行配置。
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- 创建表空间
CREATE TABLESPACE logminer_tbs DATAFILE '/home/oracle/logminer_tbs.dbf' SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

-- 创建用户family绑定表空间LOGMINER_TBS
CREATE USER family IDENTIFIED BY zyhcdc DEFAULT TABLESPACE LOGMINER_TBS QUOTA UNLIMITED ON LOGMINER_TBS;

-- 授予family用户dba的权限
 grant connect,resource,dba to family;

-- 并授予权限
  GRANT CREATE SESSION TO family;
  GRANT SELECT ON V_$DATABASE to family;
  GRANT FLASHBACK ANY TABLE TO family;
  GRANT SELECT ANY TABLE TO family;
  GRANT SELECT_CATALOG_ROLE TO family;
  GRANT EXECUTE_CATALOG_ROLE TO family;
  GRANT SELECT ANY TRANSACTION TO family;
  GRANT EXECUTE ON SYS.DBMS_LOGMNR TO family;
  GRANT SELECT ON V_$LOGMNR_CONTENTS TO family;
  GRANT CREATE TABLE TO family;
  GRANT LOCK ANY TABLE TO family;
  GRANT ALTER ANY TABLE TO family;
  GRANT CREATE SEQUENCE TO family;

  GRANT EXECUTE ON DBMS_LOGMNR TO family;
  GRANT EXECUTE ON DBMS_LOGMNR_D TO family;

  GRANT SELECT ON V_$LOG TO family;
  GRANT SELECT ON V_$LOG_HISTORY TO family;
  GRANT SELECT ON V_$LOGMNR_LOGS TO family;
  GRANT SELECT ON V_$LOGMNR_CONTENTS TO family;
  GRANT SELECT ON V_$LOGMNR_PARAMETERS TO family;
  GRANT SELECT ON V_$LOGFILE TO family;
  GRANT SELECT ON V_$ARCHIVED_LOG TO family;
  GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO family;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

本地使用Navcat连接oracle:

-- 创建 STUDENT_INFO 表
create table student_info (
  sid         number(10) constraint pk_sid primary key,
  sname       varchar2(10),
  sex         varchar2(2)
);

-- 修改STUDENT_INFO表让其支持增量日志，这句先在Oracle里创建user表再执行
ALTER TABLE FAMILY.STUDENT_INFO ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
1
2
3
4
5
6
7
8
9

2、flink cdc程序开发

maven依赖:

	
        1.8
        ${java.version}
        ${java.version}
        1.13.0
        2.12
    
    

        
            org.apache.flink
            flink-table-planner-blink_${scala.version}
            ${flink.version}
        

        
            com.ververica
            flink-connector-oracle-cdc
            2.1.0
        

        
            com.oracle.database.jdbc
            ojdbc10
            19.10.0.0
        
        
            org.apache.flink
            flink-streaming-java_${scala.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-clients_${scala.version}
            ${flink.version}
        
        
            org.apache.flink
            flink-cep_${scala.version}
            ${flink.version}
        

        
            org.apache.flink
            flink-java
            ${flink.version}
        
        
            org.apache.flink
            flink-json
            ${flink.version}
        
        
            com.alibaba
            fastjson
            1.2.68
        

        
            org.projectlombok
            lombok
            1.18.20
        
        
        
            org.slf4j
            slf4j-api
            1.7.25
        
        
            org.slf4j
            slf4j-log4j12
            1.7.25
        
        
            org.apache.logging.log4j
            log4j-to-slf4j
            2.14.0
        
    
    
        
            
                org.apache.maven.plugins
                maven-assembly-plugin
                3.0.0
                
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single

                        
                    
                
            
        
    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

功能代码:

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 测试 flink cdc 实时获取oracle数据变化
 * @author zyh
 */
public class FlinkCdcOracleExample {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.disableOperatorChaining();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE student_info (
" +
                "     SNO INT NOT NULL,
" +                   // 注意字段名要大写
                "     SNAME STRING,
" +
                "     SEX STRING,
" +
                "     PRIMARY KEY(SNO) NOT ENFORCED
" +
                "     ) WITH (
" +
                "     'connector' = 'oracle-cdc',
" +
                "     'hostname' = 'localhost',
" +
                "     'port' = '1521',
" +
                "     'username' = 'family',
" +
                "     'password' = 'zyhcdc',
" +
                "     'database-name' = 'helowin',
" +
                "     'schema-name' = 'FAMILY',
" +           // 注意这里要大写
                "     'table-name' = 'STUDENT_INFO',
" +
                "     'debezium.log.mining.continuous.mine'='true',
"+
                "     'debezium.log.mining.strategy'='online_catalog',
" +
                "     'debezium.database.tablename.case.insensitive'='false',
"+
                "     'scan.startup.mode' = 'initial')");

        TableResult tableResult = tableEnv.executeSql("select * from student_info");
        tableResult.print();
        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

启动测试:

由于设置了’scan.startup.mode’ = ‘initial’，所以程序会初始化表中现有的数据。
现在对数据表执行以下操作:

-- 新增一条数据
insert into student_info (sno, sname, sex)
values (28, 'zyh-test', 'm');
-- 更新数据
update student_info t set t.sname='zyh666', t.sex='m' where t.sno=26;
-- 删除数据
delete from student_info where sno = 25;
1
2
3
4
5
6
7

程序执行结果:

自此，oracle-cdc的功能已经测试成功，将对应的结果写到mysql表中，可以使用flink sql将对应的结果写入mysql中，由于上面执行的结果属于撤回流，需要在flink sql中创建mysql sink表的时候指定主键，写入mysql的功能这里就不再贴出，小伙伴可以自己下去实现。

3、中间遇到的问题，排查解决

1、读取数据有延迟，在create语句配置以下两个选项进行解决:

'debezium.log.mining.strategy'='online_catalog',
'debezium.log.mining.continuous.mine'='true'
1
2

2、找不到表

[ERROR] Could not execute SQL statement. Reason:
io.debezium.DebeziumException: Supplemental logging not configured for table FAMILY.STUDENT_INFO  Use command: ALTER TABLE LIUYUN.flink ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS
1
2

参看文档： https://docs.oracle.com/cd/E11882_01/server.112/e41084/sql_elements008.htm
可以在 create 语句中加上 ：

'debezium.database.tablename.case.insensitive'='false'
1

参考资料

flink oracle cdc 官方文档

另外，除此之外，还可以使用kafka connect结合debezium的方式采集oracle数据，不过使用这种方式采集的是最原始的oracle变更日志数据，需要开发者再去进行一次日志内容的解析，增加了开发的难度以及维护成本，这种方式的具体操作步骤将会在下篇文章给出，希望感兴趣的小伙伴能够点个关注，哈哈～

先自我介绍一下，小编13年上师交大毕业，曾经在小公司待过，去过华为OPPO等大厂，18年进入阿里，直到现在。深知大多数初中级java工程师，想要升技能，往往是需要自己摸索成长或是报班学习，但对于培训机构动则近万元的学费，着实压力不小。自己不成体系的自学效率很低又漫长，而且容易碰到天花板技术停止不前。因此我收集了一份《java开发全套学习资料》送给大家，初衷也很简单，就是希望帮助到想自学又不知道该从何学起的朋友，同时减轻大家的负担。添加下方名片，即可获取全套学习资料哦