Flink CDC 基于Oracle log archiving 实时同步Oracle表到Mysql（无主键） |
您所在的位置：网站首页 › 引体向上带手套好吗 › Flink CDC 基于Oracle log archiving 实时同步Oracle表到Mysql（无主键） |
|
环境说明： flink 1.15.2 Oracle 版本：Oracle Database 11g Enterprise Edition Release 11.2.0.1.0 - 64bit Production mysql 版本：5.7 windows11 IDEA 本地运行 具体环境设置和maven依赖请看上篇：Flink CDC 基于Oracle log archiving 实时同步Oracle表到Mysql_彩虹豆的博客-CSDN博客 现在操作的是源表和目标表都无主键数据实时同步，其实是在flink建立和sink表的关联时，指定了主键，只不过是flink实时同步程序的需要，sink目标表没有主键。 1. Oracel建表 CREATE TABLE "USER" ( ID NUMBER, USERNAME VARCHAR2(255), PASSWORD VARCHAR2(255)); ALTER TABLE "USER" ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS; 2. Mysql 建表 CREATE TABLE user_new ( id int(11) NOT NULL, username varchar(255) DEFAULT NULL, password varchar(255) DEFAULT NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; 3.demo如下： import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class OracleCdcToMysqlNonePrimaryKey { public static void main(String[] args) { //1.获取stream的执行环境 StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment(); senv.setParallelism(1); //2.创建表执行环境 StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv); String sourceTable = "CREATE TABLE oracle_cdc_source " + "( ID INT, " + "USERNAME STRING, " + "PASSWORD STRING" + // ", PRIMARY KEY(ID) NOT ENFORCED) " + ")WITH (\n" + "'connector' = 'oracle-cdc',\n" + "'hostname' = '1.1.1.1',\n" + "'port' = '1521',\n" + "'username' = 'flinkcdcuser',\n" + "'password' = 'flinkpw',\n" + "'database-name' = 'LMDB',\n" +//select name from v$database; "'schema-name' = 'TEST',\n" +//select SYS_CONTEXT('USERENV','CURRENT_SCHEMA') CURRENT_SCHEMA from dual; "'debezium.snapshot.mode' = 'initial',\n" + //snapshot.mode = initial 快照包括捕获表的结构和数据。指定此值将用捕获表中数据的完整表示填充主题。 //snapshot.mode = schema_only 快照只包含捕获表的结构。如果希望连接器仅捕获快照之后发生的更改的数据，请指定此值。 "'scan.incremental.snapshot.enabled' = 'false',\n" + //scan.incremental.snapshot.enabled 增量快照是一种读取表快照的新机制。增量快照与旧的快照机制相比有很多优点，包括: // (1)在快照读取期间源可以并行;(2)在快照读取期间源可以在块粒度上执行检查点;(3)在快照读取之前源不需要获取ROW SHARE MODE锁。 "'scan.incremental.snapshot.chunk.key-column' = 'ID' ,\n" + //scan.incremental.snapshot.chunk.key-column 可以指定某一列作为快照阶段切分分片的切分列。无主键表必填，选择的列必须是非空类型（NOT NULL）。 //有主键的表为选填，仅支持从主键中选择一列。 // "'scan.incremental.snapshot.chunk.size' = '8096' ,\n" + //表快照的块大小(行数)，当读取表快照时，捕获的表被分割成多个块。 "'scan.snapshot.fetch.size' = '1024',\n" + //读取表快照时每个轮询的最大读取大小。 "'connect.max-retries' = '3',\n" + //连接器应该重试构建Oracle数据库服务器连接的最大重试次数。 "'connection.pool.size'= '20',\n" + //连接池大小 "'debezium.log.mining.strategy' = 'online_catalog',\n" + //online_catalog -使用数据库的当前数据字典来解析对象id，并且不向在线重做日志中写入任何额外的信息。 // 这使得LogMiner的挖掘速度大大提高，但代价是无法跟踪DDL的变化。如果捕获的表模式很少或从不更改，那么这是理想的选择。 "'debezium.log.mining.archive.destination.name' = 'log_archive_dest_1',\n" + "'debezium.log.mining.continuous.mine'='true'," + " 'table-name' = 'USER'\n" + ")"; tEnv.executeSql(sourceTable); // tEnv.executeSql("select * from oracle_cdc_source").print();//做步骤测试使用，能正常捕获日志操作就可以打印出来，能正常打印出来，注释掉，再往下插表 String sinkTable = "CREATE TABLE mysql_cdc_sink (" + " ID INT,\n" + " USERNAME STRING,\n" + " PASSWORD STRING,\n" + "PRIMARY KEY(ID,USERNAME,PASSWORD) NOT ENFORCED\n" + ") WITH (\n" + "'connector' = 'jdbc',\n" + "'driver' = 'com.mysql.cj.jdbc.Driver',\n" + "'url' = 'jdbc:mysql://localhost:3306/test_cdc?rewriteBatchedStatements=true',\n" + "'username' = 'root',\n" + "'password' = 'root',\n"+ " 'table-name' = 'user_new',\n" + " 'connection.max-retry-timeout' = '60s'\n" + ")"; tEnv.executeSql(sinkTable); tEnv.executeSql("insert into mysql_cdc_sink select ID,USERNAME,PASSWORD from oracle_cdc_source"); } }本地测试是没有问题的，比起mysql稍微有点延迟， "'debezium.log.mining.strategy' = 'online_catalog',\n" + //online_catalog -使用数据库的当前数据字典来解析对象id，并且不向在线重做日志中写入任何额外的信息。 // 这使得LogMiner的挖掘速度大大提高，但代价是无法跟踪DDL的变化。如果捕获的表模式很少或从不更改，那么这是理想的选择。 "'debezium.log.mining.continuous.mine'='true'," +加上这俩个参数后，延迟应该就是1-3秒，如果不加的话，那就是几分钟的延迟了。 由于无主键， debezium.snapshot.mode' = 'initial',这个参数会导致，程序运行几次，源表数据就会同步几次到目标表，并不会去重，如果想一直这个参数运行，需要在插入前先清空表，但是如果是数据量大的，推荐还是先用这个参数同步历史数据，完成后，再改为 schema_only，启动程序，然后把上面一个程序干掉。 上面设置的主键是三个字段，ID、USERNAME、PASSWORD,这三个字段不能为null，如果有数据为null,程序在启动的时候，就会报错，虽然没有打印到控制台上，但是可以看到控制台程序结束了，不是一直在运行，并且数据也是同步不过去的。所以挑选主键字段时一定要确定此字段一定不为null，如果为null的话，就需要能接受转换处理，比如：varchar 类型 将null值转换为空字符串 insert into mysql_cdc_sink select case when ID is null then 0 else ID end,case when USERNAME is null then '' else USERNAME end,case when PASSWOED is null then '' else PASSWOED end from oracle_cdc_source 具体如何处理，还看业务需求。不过，在数据同步时，尽量要做到不对数据做任何变动。如果是可以加入清洗，那就随便玩。 使用记录一：源表4个字段：a,b,c,d 目标表也是一样，主键设置为 a,b,c 当在源表修改一条记录 a,b,c 字段值不动，d修改。 目标表会新增一条 a,b,c, d(修改后的值)。原记录不改变。多了一条。 在此基础上在此条记录上，源表再修改a,b,c 其中一个主键的值，d不动。 目标表会将俩条记录合并，成一条，数据为源表修改后的数据。 使用记录二：源表4个字段：a,b,c,d 目标表也是一样，主键设置为 a,b,c 当在源表新增一条记录 a,b,c 字段值和已有的一条记录(咱们先叫做源数据记录，后面会用到)保持一致，d随便。 目标表会新增一条 a,b,c, d(随便)。原记录不改变。多了一条。并不会根据主键a,b,c去重 在此基础上在此条记录上，源表删除新增的这一条数据。 目标表会新增一条源数据记录，数据变成了三条，源表删除的记录并不会同步删除。 在此基础上，源表还有一条源数据记录，修改a,b,c 其中的一个字段值，d不动。 目标表就会把新增的一条数据和源数据记录数据都删掉，和源表保持一致了。 挺有意思哈，感觉已经不能这样再摸索了，已经失控了，还是得扒扒源码一探究竟。 |
| CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |