Flink CDC实战之Oracle同步Mysql

简介

最近有个项目要对接ZF的数据，数据库为Oracle，由于采用OGG的方式同步费用很高，于是想到采用FlinkCDC的方式进行同步。

Install Oracle

本例采用docker安装oracle

1. 拉取oracle镜像

docker pull registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g

2. 创建容器

docker run -d -p 1521:1521 --name oracle11g registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g

3. 进入镜像进行配置

docker exec -it oracle11g bash
su root // 输入密码：helowin

4. 编辑profile文件配置ORACLE环境变量

export ORACLE_HOME=/home/oracle/app/oracle/product/11.2.0/dbhome_2
export ORACLE_SID=helowin
export PATH=$ORACLE_HOME/bin:$PATH

• 打开：vi /etc/profile ，在文件最后补充以上内容；
• 保存后执行source /etc/profile 加载环境变量；
• 创建软连接：ln -s $ORACLE_HOME/bin/sqlplus /usr/bin
• 切换到oracle 用户：su - oracle

5. 登录sqlplus并修改system用户密码

sqlplus /nolog   --登录
conn /as sysdba  --

alter user system identified by system;--修改system用户账号密码；
ALTER PROFILE DEFAULT LIMIT PASSWORD_LIFE_TIME UNLIMITED; --修改密码规则策略为密码永不过期；（会出现坑，后面讲解）
alter system set processes=1000 scope=spfile; --修改数据库最大连接数据；

shutdown immediate; --关闭数据库
startup; --启动数据库
exit：退出软链接

Setup Oracle

1. Enable log archiving

alter system set db_recovery_file_dest_size = 10G;
alter system set db_recovery_file_dest = '/opt/oracle/oradata/recovery_area' scope=spfile;
shutdown immediate;
startup mount;
alter database archivelog;
alter database open;

2. Check whether log archiving is enabled

-- Should now "Database log mode: Archive Mode"
archive log list;

3. Enable supplemental logging

-- Enable supplemental logging for a specific table:
ALTER TABLE SCOTT.EMP ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

-- Enable supplemental logging for database
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

我在代码中调试的时候发现在表上设置不生效，然后在库上设置之后可以了

-- nabling Supplemental Logging at Database Level
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
alter database add supplemental log data (all) columns;

Code

1. 依赖库

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-oracle-cdc</artifactId>
    <version>2.2.1</version>
</dependency>

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-sql-connector-oracle-cdc</artifactId>
    <version>2.2.1</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>connect-api</artifactId>
    <version>2.7.0</version>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.7.0</version>
</dependency>

注意：不要添加flink-connector-mongodb-cdc依赖，否则会有debezium相关的jar冲突

2. DataStream代码

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.oracle.OracleSource;

public class OracleSourceExample {
    public static void main(String[] args) throws Exception {
        SourceFunction<String> sourceFunction = OracleSource.<String>builder()
                .hostname("localhost")
                .port(1521)
                .database("HELOWIN") // monitor HELOWIN database
                .schemaList("SCOTT") // monitor SCOTT schema
                .tableList("SCOTT.EMP") // monitor EMP table
                .username("system")
                .password("system")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(sourceFunction).print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute();
    }
}

3. Flink SQL代码

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.util.concurrent.TimeUnit;

public class FlinkCdcOracleSync {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setBufferTimeout(200);
        env.enableCheckpointing(5 * 60 * 1000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(4 * 60 * 1000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30 * 1000);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.of(5, TimeUnit.MINUTES),Time.of(10, TimeUnit.SECONDS)));
        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("CREATE TABLE oracle_emp_source (" +
                "  EMPNO INTEGER," +
                "  ENAME STRING," +
                "  JOB STRING," +
                "  MGR STRING," +
                "  HIREDATE TIMESTAMP," +
                "  SAL INTEGER," +
                "  COMM INTEGER," +
                "  DEPTNO INTEGER," +
                "  PRIMARY KEY(EMPNO) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'oracle-cdc'," +
                "  'hostname' = 'localhost'," +
                "  'port' = '1521'," +
                "  'username' = 'system'," +
                "  'password' = 'system'," +
                "  'database-name' = 'HELOWIN'," +
                "  'schema-name' = 'SCOTT'," +
                "  'table-name' = 'EMP'" +
                ")");

        tableEnv.executeSql("CREATE TABLE mysql_emp_sink (" +
                "  EMPNO INTEGER," +
                "  ENAME STRING," +
                "  JOB STRING," +
                "  MGR STRING," +
                "  HIREDATE TIMESTAMP," +
                "  SAL INTEGER," +
                "  COMM INTEGER," +
                "  DEPTNO INTEGER," +
                "  PRIMARY KEY(EMPNO) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'jdbc'," +
                "  'url' = 'jdbc:mysql://127.0.0.1:3306/wlapp'," +
                "  'table-name' = 'EMP'," +
                "  'driver' = 'com.mysql.jdbc.Driver'," +
                "  'username' = 'root'," +
                "  'password' = '123456'," +
                "  'scan.fetch-size' = '200'" +
                ")");

        tableEnv.executeSql("insert into mysql_emp_sink select * from oracle_emp_source");
    }
}

结

调试过程中重启了一次电脑，然后重启oracle容器的时候发现oracle死活连不上了，后面索性删除容器重新建了一个，问题解决了。本例只是简单做了下flink cdc oracle的测试，后续整库同步还需要继续研究。