简介
最近有个项目要对接ZF的数据，数据库为Oracle，由于采用OGG的方式同步费用很高，于是想到采用FlinkCDC的方式进行同步。
Install Oracle
本例采用docker安装oracle
- 拉取oracle镜像
docker pull registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g
- 创建容器
docker run -d -p 1521:1521 --name oracle11g registry.cn-hangzhou.aliyuncs.com/helowin/oracle_11g
- 进入镜像进行配置
docker exec -it oracle11g bash
su root // 输入密码：helowin
- 编辑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
- 登录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
- 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;
- Check whether log archiving is enabled
-- Should now "Database log mode: Archive Mode"
archive log list;
- 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
- 依赖库
<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冲突
- 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();
}
}
- 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的测试，后续整库同步还需要继续研究。
