Oracle 使用DB_Link同步MySQL数据

一、创建DB_LINK1、Oracle验证MySQL驱动注意Oracle数据库在安装过程中会自动安装MySQL的驱动“dg4odbc”。目前仅需要验证该驱动即可。当前环境中存在该驱动。$file$ORACLE_HOME/bin/dg4odbc2、配置驱动连接注意此处需要在Oracle数据库中配置驱动连接文件。同时MySQL数据库是严重区分大小写需要注意驱动文件中的大小敏感。$vi$ORACLE_BASE/odbc.ini[myodbc5]Driver/home/dbs/app/myodbc-x.x.x/lib/libmyodbc5.so DescriptionConnector/ODBC x.x Driver DSN SERVERIP ADDRESSPORTMYSQL PORT NUMBERUSERmysql_user PASSWORD***** DATABASEtestOPTION0TRACEOFF3、配置监听注意此处需要在listener.ora和tnsnames.ora文件中均需要配置MYSQL监听。# tnsname.ora文件$vi$ORACLE_HOME/rdbms/admin/[SID]/tnsnames.ora myodbc5(DESCRIPTION(ADDRESS(PROTOCOLTCP)(HOST[MYSQL HOST])(PORT1521))(CONNECT_DATA(SIDmyodbc5))(HSOK))# listener.ora文件$vi$ORACLE_HOME/network/admin/listener.oraSID_LIST_LISTENER(SID_LIST(SID_DESC(SID_NAMEmyodbc5)(ORACLE_HOME/home/dbs/app/Ora/product/11.2.0/dbhome_1)(PROGRAMdg4odbc)(ENVLD_LIBRARY_PATH/home/dbs/app/unixodbc-2.2.14/lib:/home/dbs/app/Ora/product/11.2.0/dbhome_1/lib)))4、配置透明网关注意此处网关为数据库的不是OS的网关$vi$ORACLE_HOME/hs/admin/initmyodbc5.oraHS_FDS_CONNECT_INFOmyodbc5# Data source name in odbc.iniHS_FDS_TRACE_LEVELOFF -- 关闭追踪降低负载如出现错误可以开启追踪日志文件在$ORACLE_HOME/hs/ 下HS_FDS_SHAREABLE_NAME/home/dbs/app/unixodbc-2.3/lib/libodbc.so ß 系统中该文件的路径HS_FDS_SUPPORT_STATISTICSFALSEHS_LANGUAGEAMERICAN_ AMERICA. AL32UTF8 AMERICAN_AMERICA.WE8ISO8859P15ß Oracle建议当前值但也可以输入Oracle字符集原因是ODBC可能不识别字符集## ODBC env variablessetODBCINI/home/dbs/etc/odbc.ini5、重载监听$ lsnrctl reload – 重新加载6、创建DB_LINK注意myodbc5是tnsnames.ora文件中的监听服务名SQLcreate public databaselinkOM_LINK connect toXXXXidentified byXXXXXXusingmyodbc5;7、测试连接即可 SQLselect* fromtab1myodbc5;二、新建同步测试案例1、建立同步表Oracle数据库注意需要从MySQL同步数据过来建立对应的字段并且设置好数据类型即可。当前MySQL和Oracle的字符集分别是“utf8mb4”和“AMERICAN_ AMERICA. AL32UTF8”SQLcreate table test_t1(id integer,create_datedate);2、新建同步记录表注意因为数据是不断变化的在做增量同步时需要考虑一个关键值且该值时有规律的变化或者主键ID、时间戳都可以。当前经了解其ID是使用系统自增的那么可以通过ID来判断哪些数据是新增的将最后同步的最大的ID写到同步记录表中下次自动同步时获取该值与MySQL数据库中的表进行比对比该值小的及同步到Oracle数据库中。create table om_sync_control(table_name varchar2(50)primary key, -- 被同步的表名 last_sync_id number(18), -- 上次同步的最大ID last_sync_time date, -- 上次同步时间 sync_status varchar2(10)-- 状态SUCCESS/FAILED);3、创建错误信息表注意该表用来记录每次同步数据时是否发生报错如果出现报错将记录对应的报错信息CREATE TABLE om_sync_error_log(table_name VARCHAR2(50), -- 被同步的表名 ids NUMBER(20), -- 失败的ID om_error BLOB, -- 错误原因 insert_date DATE -- 插入日期);4、初始化同步记录表SQLinsert into om_sync_control(table_name, last_sync_id, last_sync_time, sync_status)values(test_t1,0, sysdate,SUCCESS);5、创建存储过程-- 创建同步数据存储过程 create or replace procedure SYNC_ON_INCREMENTAL is v_max_id number :0;-- 记录最大值 v_count number :0;-- 插入数据总量 v_last_sync_id number;-- 上次同步的最大值 v_sync_timedate:sysdate;-- 同步时间 -- 自定义异常 e_sync_error exception -- 游标查询上次同步后新增的数据 cursor new_data_cur isselectid, create_date from test_t1OM_LINK whereid(select nvl(last_sync_id,0)from om_sync_control where table_nametest_t1)order byid;-- 按ID排序确保顺序处理 begin -- 获取当前已同步的最大ID用于后续更新 SELECT nvl(last_sync_id,0)INTO v_last_sync_id FROM om_sync_control WHERE table_nametest_t1FOR UPDATE;-- 加锁防止并发同步 dbms_output.put_line(开始增量同步上次同步ID||v_last_sync_id);-- 遍历新增数据forrecinnew_data_cur loop begin -- 插入到Oracle目标表 insert into test_t2(id, create_date)values(rec.id, -- 使用MySQL的自增ID rec.create_date);-- 记录本次循环的最大IDifrec.idv_max_idthenv_max_id :rec.id;endif;v_count :v_count 1;-- 每100条提交一次避免大事务ifmod(v_count,100)0thencommit;dbms_output.put_line(已同步 ||v_count|| 条当前最大ID||v_max_id);endif;exception when othersthen-- 记录错误信息继续处理下一条 dbms_output.put_line(插入失败ID||rec.id||错误||sqlerrm);--insertinto om_sync_error_log values(test_t1,rec.id,error,sysdate);insert into om_sync_error_log values(test_t1,rec.id,sqlerrm,sysdate);commit;end;end loop;-- 更新同步记录表ifv_max_id0thenupdate om_sync_controlsetlast_sync_idv_max_id, last_sync_timev_sync_time, sync_statusSUCCESSwhere table_nametest_t1;commit;dbms_output.put_line(增量同步完成。共同步 ||v_count|| 条新数据最大ID||v_max_id);elsedbms_output.put_line(没有新数据需要同步。);endif;exception when othersthenrollback;dbms_output.put_line(同步过程出错||sqlerrm);-- 更新状态为失败 update om_sync_controlsetsync_statusFAILED, last_sync_timesysdate where table_nametest_t1;commit;raise;end SYNC_ON_INCREMENTAL;6、配置定时任务-- 开启定时任务 BEGIN dbms_scheduler.create_job(job_nameSYNC_ON_INCREMENTAL_FIRST, --定时任务名称 job_typeSTORED_PROCEDURE,--类型为存储过程job_actionSYNC_ON_INCREMENTAL,--存储过程名称 start_datesysdate, --首次执行时间repeat_intervalFREQHOURLY;INTERVAL12, --十二小时执行一次 --【FREQminutely;表示间隔单位可选值Yearly,monthly,weekly,datly,hourly,minutely,secondly】enabledtrue -- 立即启用定时任务);END;7、查看定时任务是否启动ENABLETRUE OR FALSE标志定时任务是否启动STATERUNNING标志定时任务运行中、SCHEDULED定时中-- 查询定时任务是否启动 Select * from dba_scheduler_jobs where job_nameSYNC_ON_INCREMENTAL_FIRSTorder by start_date desc;8、查看定时任务具体执行信息STATUSSUCCEEDED和FAILED定时任务执行成功或失败REQ_START_DTAE定时任务执行时间ERRORS记录系统报错信息如违反唯一约束条件、程序包或函数无效等OUTPUT输出存储过程里面的结束时定义的提示语select* from dba_scheduler_job_run_details where job_nameSYNC_ON_INCREMENTAL_FIRSTorder by log_date desc;9、关闭定时任务begin dbms_scheduler.drop_job(SYNC_ON_INCREMENTAL_FIRST,forcetrue);-- 删除定时任务force无论当前定时任务什么状态都删除 --dbms_scheduler.run_job(TEST_JOB4,true);-- 手动运行定时任务 --dbms_scheduler.enable(TEST02);-- 启用定时任务根据定义的执行频率Oracle会自动决定什么时候开始运行 --dbms_scheduler.disable(TEST_JOB2);end;三、案例参考注意当前案例最重要的是体现当前方案的处理思路以及结果验证。但没有使用DB_LINK。-- 创建序列 CREATE SEQUENCE test_oracle_mysql START WITH1INCREMENT BY1NOCACHE;--建立测试表 create table test_t1(id integer,create_datedate);create table test_t2(id integer,create_datedate);-- 创建同步记录表 create table om_sync_control(table_name varchar2(50)primary key, -- 被同步的表名 last_sync_id number(18), -- 上次同步的最大ID last_sync_time date, -- 上次同步时间 sync_status varchar2(10)-- 状态SUCCESS/FAILED);select* from om_sync_error_log;-- 创建错误信息表 CREATE TABLE om_sync_error_log(table_name VARCHAR2(50), -- 被同步的表名 ids NUMBER(20), -- 失败的ID om_error BLOB, -- 错误原因 insert_date DATE -- 插入日期);drop table om_sync_error_log;-- 初始化记录第一次同步前执行 insert into om_sync_control(table_name, last_sync_id, last_sync_time, sync_status)values(test_t1,0, sysdate,SUCCESS);--建立原表存储过程插入测试数据 create or replace procedure test_om is begin--insertinto test_om_consumselect* from consum_ae_test;insert into test_t1(id,create_date)values(test_oracle_mysql.nextval,sysdate);commit;end test_om;-- 定时增加原表测试数据 BEGIN dbms_scheduler.create_job(job_nametest_job_om, --定时任务名称 job_typeSTORED_PROCEDURE,-- 类型为存储过程job_actiontest_om,-- 存储过程名称 start_datesysdate, --执行时间--end_dateSYSDATE 3/1440,--repeat_intervalFREQMINUTELY;INTERVAL1, --可选执行频率为空时表示只执行一次 --【FREQminutely;表示间隔单位可选值Yearly,monthly,weekly,datly,hourly,minutely,secondly】enabledtrue);-- 立即执行 END;-- 数据查询select* from test_t1 order byiddesc;select* from test_t2 order byiddesc;selectmax(id)from test_t1;selectmax(id)from test_t2;select* from om_sync_control;select* from dba_db_links;-- 创建同步数据存储过程目前暂未使用db-link先把整个流程跑通仅缺少新建DB-LINK create or replace procedure SYNC_ON_INCREMENTAL is v_max_id number :0;-- 记录最大值 v_count number :0;-- 插入数据总量 v_last_sync_id number;-- 上次同步的最大值 v_sync_timedate:sysdate;-- 同步时间 -- 自定义异常 e_sync_error exception;-- 游标查询上次同步后新增的数据 cursor new_data_cur isselectid, create_date from test_t1 whereid(select nvl(last_sync_id,0)from om_sync_control where table_nametest_t1)order byid;-- 按ID排序确保顺序处理 begin -- 获取当前已同步的最大ID用于后续更新 SELECT nvl(last_sync_id,0)INTO v_last_sync_id FROM om_sync_control WHERE table_nametest_t1FOR UPDATE;-- 加锁防止并发同步 dbms_output.put_line(开始增量同步上次同步ID||v_last_sync_id);-- 遍历新增数据forrecinnew_data_cur loop begin -- 插入到Oracle目标表 insert into test_t2(id, create_date)values(rec.id, -- 使用MySQL的自增ID rec.create_date);-- 记录本次循环的最大IDifrec.idv_max_idthenv_max_id :rec.id;endif;v_count :v_count 1;-- 每100条提交一次避免大事务ifmod(v_count,100)0thencommit;dbms_output.put_line(已同步 ||v_count|| 条当前最大ID||v_max_id);endif;exception when othersthen-- 记录错误信息继续处理下一条 dbms_output.put_line(插入失败ID||rec.id||错误||sqlerrm);--insertinto om_sync_error_log values(test_t1,rec.id,error,sysdate);insert into om_sync_error_log values(test_t1,rec.id,error,sysdate);commit;end;end loop;-- 更新同步记录表ifv_max_id0thenupdate om_sync_controlsetlast_sync_idv_max_id, last_sync_timev_sync_time, sync_statusSUCCESSwhere table_nametest_t1;commit;dbms_output.put_line(增量同步完成。共同步 ||v_count|| 条新数据最大ID||v_max_id);elsedbms_output.put_line(没有新数据需要同步。);endif;exception when othersthenrollback;dbms_output.put_line(同步过程出错||sqlerrm);-- 更新状态为失败 update om_sync_controlsetsync_statusFAILED, last_sync_timesysdate where table_nametest_t1;commit;raise;end SYNC_ON_INCREMENTAL;-- 开启定时任务 BEGIN dbms_scheduler.create_job(job_nameSYNC_ON_INCREMENTAL_FIRST, --必选 任务名称 job_typeSTORED_PROCEDURE,--必须 任务类型【存储过程匿名块等】job_actionSYNC_ON_INCREMENTAL,--必须 任务内容与job_type配合使用 start_datesysdate, --可选 首次执行时间为空时表示立即执行 -- end_dateSYSDATE 5/1440,repeat_intervalFREQMINUTELY;INTERVAL2, --可选执行频率为空时表示只执行一次 --【FREQminutely;表示间隔单位可选值Yearly,monthly,weekly,datly,hourly,minutely,secondly】enabledtrue);--可选 是否启用任务,注意这里是enabled;END;-- 查询定时任务执行具体信息select* from dba_scheduler_job_run_details where job_nameSYNC_ON_INCREMENTAL_FIRSTorder by log_date desc;-- 查询定时任务是否启动select* from dba_scheduler_jobs where job_nameSYNC_ON_INCREMENTAL_FIRSTorder by start_date desc;-- 删除、运行、启动、关闭定时任务 begin dbms_scheduler.drop_job(SYNC_ON_INCREMENTAL_FIRST,forcetrue);-- 删除定时任务force无论当前定时任务什么状态都删除 --dbms_scheduler.run_job( SYNC_ON_INCREMENTAL_FIRST ,true);-- 手动运行定时任务 --dbms_scheduler.enable( SYNC_ON_INCREMENTAL_FIRST );-- 启用定时任务根据定义的执行频率Oracle会自动决定什么时候开始运行 --dbms_scheduler.disable( SYNC_ON_INCREMENTAL_FIRST );end;