1、SymmetricDS數據同步工具安裝使用說明書隨著大數據產品功能的豐富，以及用戶對于大數據的時效性的需求，需要款能夠實現跨數據庫，以及同一種數據庫跨版本的數據同步工具，同時要能夠支撐數據庫到Kafka的數據推送。目前大數據技術組研發了一款yhbi的同步工具，支持從oracle到oracle的同步，以及oracle到Kafka的同步，但是受限于不能支持oracle12c版本數據的同步。OGG也可以實現oracle至ijoracle,以及oracle到其他目標端數據庫或者消息隊列的同步，但OGG的費用、維護成本較高。故采用了一款名為SymmetricDS的開源數據庫同步工具。2.2.簡介Symm

2、etricDS就像其名稱一樣，是為了實現數據源的對稱性也就是數據同步。SymmetricDS是用于數據庫和文件同步的開源軟件，支持多主復制、過濾同步和轉換。它使用web（HTTP傳輸）和數據庫技術（觸發器）將更改數據復制為預定的或接近實時的操作，并且它還包含了完整數據負載的初始負載特性。該軟件的設計目的是針對大量節點，跨低帶寬連接工作，并經得起網絡中斷。2.1.概述SymmetricDS的同步原理如下圖所示。節點負責使用HTTP將來自數據庫或文件系統的數據與網絡中的其他節點進行同步。節點被分配給作為一個單元配置在一起的節點組之一。節點組與組鏈接鏈接在一起，以定義推或拉通信。pull使一個節點與

3、其他節點連接并請求正在等待的更改，而push使一個節點在需要發送更改時與其他節點連接。每個節點都通過使用連接URL用戶名和密碼的Java數據庫連接（JDB砥區動程序連接到數據庫。雖然節點可以跨廣泛的區域網絡進行分隔，但是為了獲得最佳性能，節點所連接的數據庫應該位于局域網絡附近。節點使用其數據庫連接創建表作為配置設置和運行時操作的數據模型。用戶填充配置表以定義同步，運行時表捕獲更改并跟蹤活動。要同步的表可以位于連接可訪問的任何目錄和模式中，而要同步的文件可以位于本地服務器可訪問的任何目錄中。在啟動時，SymmetricDS查找節點屬性文件，并為找到的每個文件啟動一個節點，這允許多個節點在同一個實

4、例中運行并共享資源。節點的屬性文件包含其外部ID、節點組、注冊服務器URL和數據庫連接信息。外部ID是用于從其他節點識別它的節點的名稱。一個節點被配置為存儲主配置的注冊服務器。當第一次啟動節點時，它使用發送外部ID和節點組的注冊進程與注冊服務器聯系。作為響應，節點接收其配置和節點密碼，這些密碼必須在與其他節點同步時作為身份驗證發送。2.2.架構節點中的每個子系統負責部分數據移動，并通過配置進行控制。數據流經系統的步驟如下：3.捕獲到源數據庫的運行時表中HTTPSHTTPSNodeExternalIO0JDBCJDBCHTTP5NodeLternnrIDLternnrID1 1Gtcupclie

5、ExternalID.ExternalID.2 2G Grouproupclientclient4.發送到目標節點和成批組的路由5.提取并轉換為輸出批處理所需的行、列和值6.將發出的批處理發送到目標節點7.在目標節點接收到的批處理8.轉換為進入批處理所需的行、列和值9.加載數據并向源節點返回確認信息架構圖如下圖所示:%.3.%.3.自動恢復數據傳送是持久的和低維護，承受停機時間和自動從網絡故障恢復。%.3.%.3.安全和高效通信使用為低帶寬網絡設計的數據協議，并通過HTTPS專輸加密傳輸。CaptureRouteIncomingIncomingTransformLoadSytnm&tf

6、icDECaptureHouleIncomingIncomingTraristormLoadSymmetricOGSymmetricOG特點SymmetricDSI供了豐富的特性集，具有靈活的配置，可用于在具有多個系統的混合環境中進行大規模部署。#數據同步更改關系數據庫的數據捕獲和文件系統的文件同步可以是周期性的，也可以是接近實時的，有一個初始負載特性可以完全填充節點。#集中管理從一個可以調查和解決沖突和錯誤的集中位置配置、監視和故障排除同步。JDBCJDBCExtractDalabaseTranstormFileSystemBatchOutgoingOutgoingHTTP/SExtract

7、TrariSfoTiFileSyslemFileSyslem%.3.%.3.轉換在多個點上操作數據，以過濾、子集、轉換、合并和豐富數據。%.3.%.3.沖突管理通過配置自動和手動解決的規則來加強雙向同步的一致性。%.3.%.3.可擴展可以配置腳本和Java代碼來處理事件、轉換數據和創建自定義行為。%.3.%.3.部署選項軟件可以作為獨立的自包含服務器安裝，部署到web應用服務器，或者嵌入到應用程序中。3.3.安裝SymmetricDS官方提供了3種安裝方法。獨立安裝SymmetricDS安裝和運行作為一個獨立的進程使用內置Jettyweb服務器。 這是安裝實例最簡單也是最推薦的方法。WebAr

8、chive(WAR)WebArchive(WAR)將SymmetricDSWebArchive（WA國件部署到單獨安裝、維護和運行的現有Web應用程序容器中。嵌入式SymmetricDS嵌入在現有的應用程序中。在這個選項中，編寫了一個定制的包裝程序，它調用SymmetricDSAP來同步數據。在本文檔中，只對前兩種方法做介紹，并且是在同一臺服務器上部署corp節點（主）和store節點（從），實現一主一從的數據同步。#安裝前準備3 3下載下載地址：https:/ 3上傳把文件上傳到服務器。#獨立安裝3 3主節點unzipsymmetric-server-3.9.13.zipmvsymmetri

9、c-server-3.9.13symmetric-corp配置文件cdsymmetric-corpsample目錄下默認放置了一些配置文件，需要復制到engines目錄下。cpsamples/pertiesengines/viengines/pertiesongino.naro-*db.dnverM?irysqL.jdbc.OrivardbH(jrirdO.dNerojwstqesqL(jver#db.drLer=or;|apache,derb/.jdbcHEhbeddedDriver#db.driver-urj.hsqldb.jdbcDrive

10、r如七.dnv&rnat.toijrcflfonga.jtdE.dbc.Driver#dtirdrivAr=cDnUmdb?,jcc.PF?Drisr?db.drier=con.mfonix.jd&c.ifxD-iver#db.dri/er-or5,fire&irdsql.dbc.FBDrivsr#ercIient.Driver*曲.duv電r=org10BC#如rdrLwer=cosybatejdbc*jrtjc.syborir#dL.山iverson.iiuuiJi.jdbc.Diivei*db.driver-or;.h

11、2.OriveraTh#JDGCURLusedto8rHMttothedatjbdE孝柚.ul=： dbc： ri產t1：】nrallot/rnTriT/rntilSPdtliLl=#db.url=jdbc；jD5tgre5ql；/7locfllhost/corriitype=jiispecified#db.jrl-dtc：/locilhost;1433iuseCur&ors=tne：butf&rRaHeni)ry=104e：lot)Ehjfter=5242380孑此.url=jdbc：lb2i/ZlocdUiUit/coip#db.url-dkc；Lnformix-sqli；

12、/localhost；9OBE/Cirp:Ifir9RMIXSERVERolids1LS01#db.url-jdkc：firbi*dEql.；ljcalho3t；var/lib/fircbir(i/data/databa5(?iaffe.url=jdbciinterbit&；/l&calho9t/；cipt/interba5#/clata/corpJgdlh*dtiPurl-jdbC!lte:cnqj.5fil.iTe#dL.urljdtx；;$yL：Tds：lin；dlhast：50CO/ildtbdenair#dbHurl-jcbc；CMrtnuadb；/IocalhD3t

13、/dat3bse?5chwidatabase善曲,urlc門2；8口；所O-SER峨R-FJE;LOCK_TIT：QUT-60(0dbHuser-db.passwrrrt=sync.u-1=qrLp.idextemat.id-;c-b.pu飛 口 .pariod上111。.尸互=rknoJtgpn門江t=用號.帖二ji-b.pubh.jerijil.lnie.iBS=j)tb.pull.periid.tine,msj-initial.lcddr:rcate.firnt-需要修改源庫連接信息db.driverdb.urldb.userdb.password同步的主節點地址sync.url端口配置

14、Specilythehostname/IPaddresstobindto.(Defaultwillbindtoallinterlaces.)#host.bind.nan&=aEnabletyfichirtinizationaverHTTP.#tiTtp.enatile=PortnuiiberfcrsynchronzatioooverHTTP.Ihttp.portaEnabletyncronizatiftnovarHTTPS1HmpvarSSL).#httfjs.eHdbl”ilPotnumberforsynchroAizationoverHTTPS(HTTPoierSSL

15、).用https.port-lMl-Ifeftatruiurljdksqlitmstn2的L岡IL2dda,url=jdbr：ybasezTcferlocalbnit:50&l/databdSE?namp客dh：cm.nucdb;/IocaIhst,database?schaia=datebase卷do.ur1=dbc；h2；5t0rl；AUTO_SERVER=rRUE:LOCK_TIMEOUT=fiMM曲.UM；工db.p9-vord-registration,uirlitrp：qaup+id,tLextemiil-nd-OD.utinq.pericdtLeBrs-jon.pugiK

16、ptirod.t-Lne.ns三jg枷ll,period,tlwag=需要修改目標庫連接信息db.driverdb.urldb.userdb.password從節點的注冊地址，與主節點的sync.url保持一致registration.url.端口配置host.birdmame=http.anable=#http.port*https.enable=PortnumberforsynchronizationoverHTTPS(HTTPoverSSL),#https+pcrtf-1Us白atrustmanagerthatalloweself-signedserverSSLcertif

17、icates.#https,allo.self.signed,certs=Listhostnamesthatareallowedfor5veSSLcertificates.*https.venfj_ed,s白廣電廣.#EnableJavaManagementExtensions(JIMXIwebconsole.#jmx.http-enable=T-I.-若PortnimberforJavaManagementExtensions(JHX)webconsole.w5|nx.httpporizk初始化主節點導入item相關表./bin/dbimport-enginecorp-000-formatX

18、ML-alter-casesamples/create_sample.xml導入系統表./bin/symadmin-enginecorp-000create-sym-tables導入item相關表的數據./bin/dbimport-enginecorp-000samples/insert_sample.sql啟動在主節點目錄./bin/sym等待主節點啟動完畢。在從節點目錄./bin/symStore-001節點啟動后， 需要等待一段時間， 主節點corp-00會把系統配置數據發送到從節點store-0010當日志信息穩定后，查詢store-001的目標數據庫的item表，如果有數據表示同步已

19、完成。測試同步修改主節點corp-000源庫的item表數據， 查詢從節點store-001目標數據庫里的item表是否有變化，如果數據一樣，表示同步成功。3.3.WebArchive使用命令進行打包。#bin/symadmin-pengine/pertiescreate-war/some/path/to/symmetric-ds.war注意事項perties文件是指的perties或者perties。打包完成后，該文件在WEB-INF/classesF面，并且名稱變為

20、了perties。打包后可發布在Tomcat,Jetty,orJBoss不支持直接發布在weblogic,需要配置weblogic.xml。一主一從需要打包兩個war包。注意修改perties里面的url地址。. .使用配置同步表接下來是配置我們需要同步的表。首先要明確我們需要同步哪些表，需要列出一個同步表的清單。假設我們需要同步AC01表。在源庫中找到AC01表的建表sql,在目標庫中執行該sql創建同步表。SYM_CHANNELSYM_CHANNELSYM_CHANNE康保存的是同步數據時的通道(管道)。在源庫中執行以下sql。INSER

21、TINTOSYM_CHANNEL(channel_id,data_event_action,create_time,last_update_by,last_update_time)VALUES(ac01,NULL,SYSTIMESTAMP,admin,SYSTIMESTAMP);SYM_TRIGGERSYM_TRIGGERSYM_TRIGGER存的是要同步的表的觸發器。在源庫中執行以下sql。insertintoSYM_TRIGGER(trigger_id,source_table_name,channel_id,last_update_time,create_time)values(ac01

22、,ac01,ac01,systimestamp,systimestamp);SYM_TRIGGER_ROUTERSYM_TRIGGER_ROUTER該表是保存的觸發器路由信息，表示從哪個節點發送到哪個節點，此處應該是從主節點到從節點。在源庫中執行以下sqloinsertintoSYM_TRIGGER_ROUTER(trigger_id,router_id,initial_load_order,create_time,last_update_time)values(ac01,corp_2_store,1,systimestamp,systimestamp);初始化負載當配置好了后，需要把源庫中表

23、的數據初始化同步到目標庫的同步表中。在主節點的命令窗口執行以下命令。#./bin/symadmin-enginecorp-000reload-node001NOTE初始化負載之前一定要先創建好通道、觸發器以及觸發器路由。取消初始化負載修改SYM_TRIGGER_ROUTER勺INITIAL_LOAD_ORDER字段值為負數，則下次再執行初始化命令時，將不會再初始化該表數據。自動建目標表自動創建表容易出現異常，暫不建議使用在SYM_TABLE_RELOAD_REQUEST插入一條t己錄，create_table設置為1,表示在執行初始化命令時，先在目標數據庫中創建表。insertintoSYM_

24、TABLE_RELOAD_REQUEST(target_node_id,source_node_id,trigger_id,router_id,create_time,create_table,last_update_time)values(corp-000,store-001,ac01,corp_2_store,current_timestamp,1,current_timestamp);KafkaSymmetricDS僅可以實現數據庫導數據庫的同步，還可以實現數據庫到kafka的同步推送。采用擴展的形式實現數據的同步推送。在源庫中執行以下腳本添加擴展代碼，即可完成數據推送。declarej

25、ava_textsym_extension.EXTENSION_TEXT%type;beginjava_text:=ducer.*;importorg.jumpmind.db.model.Table;importorg.jumpmind.symmetric.io.data.CsvData;importorg.jumpmind.symmetric.io.data.DataContext;importorg.jumpmind.symmetric.io.data.DataEventType;importorg.jumpmind.sy

26、mmetric.io.data.writer.IDatabaseWriterFilter;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importjava.util.HashMap;importjava.util.Map;/*kafka寫入過濾器*author詹祥*created2019-09-2016:09*modifiedon2019-02-2118:00添力口update的處理，包含oldDatanewData*/publicclassKafkaWriterFilterimplementsIDatabaseWriterFilt

27、erprotectedfinalStringKAKFA_TEXT_CACHE=KAKFA_TEXT_CACHE+this.hashCode();privatefinalLoggerlog=LoggerFactory.getLogger(getClass();publicbooleanbeforeWrite(DataContextcontext,Tabletable,CsvDatadata)如果是symmetricDS的系統表，則不做處理if(table.getName().toUpperCase().startsWith(SYM_)returntrue;else/(Proces

28、singtable+table+forKafka);StringoldData=null;StringrowData=data.getParsedData(CsvData.ROW_DATA);if(data.getDataEventType()=DataEventType.DELETE)rowData=data.getParsedData(CsvData.OLD_DATA);elseif(data.getDataEventType()=DataEventType.UPDATE)oldData=data.getParsedData(CsvData.OLD_DATA);StringBufferka

29、fkaText=newStringBuffer();if(context.get(KAKFA_TEXT_CACHE)!=null)kafkaText=(StringBuffer)context.get(KAKFA_TEXT_CACHE);/是否使用json格式傳輸數據booleanuseJson=true;chart=”;Characters=newCharacter(t);Stringquote=s.toString();if(useJson)/UPDATE需要原有數據，存在兩個數據頭newDataoldDataif(data.getDataEventType().toString().eq

30、uals(UPDATE)kafkaText.append(+quote+tablename+quote+:+quote).append(table.getName()+quote+,).append(quote+eventType+quote+:+quote+data.getDataEventType()+quote+,).append(quote+newData+quote+:);for(inti=0;itable.getColumnNames().length;i+)kafkaText.append(quote+table.getColumnNames()i+quote+:+quote+r

31、owDatai);if(i+1table.getColumnNames().length)kafkaText.append(quote+”,);kafkaText.append(quote+,+quote+oldData+quote+:);for(inti=0;itable.getColumnNames().length;i+)kafkaText.append(quote+table.getColumnNames()i+quote+:+quote+oldDatai);if(i+1table.getColumnNames().length)kafkaText.append(quote+,);ka

32、fkaText.append(quote+);/INSERT和DELETES接用data的數據頭elsekafkaText.append(+quote+tablename+quote+:+quote).append(table.getName()+quote+,).append(quote+eventType+quote+:+quote+data.getDataEventType()+quote+,).append(quote+data+quote+:);for(inti=0;itable.getColumnNames().length;i+)kafkaText.append(quote+ta

33、ble.getColumnNames()i+quote+:+quote+rowDatai);if(i+1table.getColumnNames().length)kafkaText.append(quote+,);kafkaText.append(quote+);elsekafkaText.append(TABLE).append(,).append(table.getName().append(,).append(EVENT).append(,).append(data.getDataEventType().append(,);for(inti=0;itable.getColumnName

34、s().length;i+)kafkaText.append(table.getColumnNames()i).append(,).append(rowDatai);if(i+1table.getColumnNames().length)kafkaText.append(,);sendKafkaMessage(kafkaText.toString();/context.put(KAKFA_TEXT_CACHE,kafkaText);returntrue;publicvoidafterWrite(DataContextcontext,Tabletable,CsvDatadata)publicbo

35、oleanhandlesMissingTable(DataContextcontext,Tabletable)returntrue;publicvoidearlyCommit(DataContextcontext)publicvoidbatchComplete(DataContextcontext)publicvoidbatchCommitted(DataContextcontext)publicvoidbatchRolledback(DataContextcontext)publicvoidsendKafkaMessage(StringkafkaText)Mapconfigs=newHash

36、Map();/公司環境/con巾gs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,“20:6667,21:6667,22:6667);/四川省就業局環境configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,31:6667,32:6667,33:6667,34:6667,35:6667);configs.put(ProducerCon巾g.VALUE_SERIALIZER_CLASS_CONFIG,“mon.serialization.StringSerializer);configs.put(ProducerCon巾g.KEY_SERIALIZER_CLASS_CONFIG,“mon.serialization.StringSerializer);configs.put(ProducerConfig.CLIENT_ID_CONF