[發(fā)明專利]一種數(shù)據(jù)采集同步系統(tǒng)與同步方法

在審

• 申請(qǐng)?zhí)枺?/td>: 202110771203.0
• 申請(qǐng)日: 2021-07-08
• 公開（公告）號(hào): CN113505173A
• 公開（公告）日: 2021-10-15
• 發(fā)明（設(shè)計(jì)）人: 許士松;朱坤奎
• 申請(qǐng)（專利權(quán)）人: 上海卓鋼鏈科技有限公司
• 主分類號(hào): G06F16/27
• 分類號(hào): G06F16/27;G06F16/22;G06F16/215;G06F16/28
• 代理公司: 上海中外企專利代理事務(wù)所(特殊普通合伙) 31387
• 代理人: 孫益青

1.一種數(shù)據(jù)采集同步系統(tǒng)，其特征在于，包含用戶端、服務(wù)端，所述服務(wù)端內(nèi)置有數(shù)據(jù)庫(kù)、審核模塊、采集同步模塊；

所述用戶端用于生成傳輸至所述服務(wù)端的全量采集同步請(qǐng)求，所述全量采集同步請(qǐng)求內(nèi)存儲(chǔ)有采集數(shù)據(jù)；

所述服務(wù)端將所述全量采集同步請(qǐng)求分別傳輸至所述數(shù)據(jù)庫(kù)、所述審核模塊；所述審核模塊用于接收所述全量采集同步請(qǐng)求并審核并生成傳輸至所述采集同步模塊、所述數(shù)據(jù)庫(kù)、所述用戶端的審核數(shù)據(jù)，所述審核數(shù)據(jù)內(nèi)存儲(chǔ)有所述采集數(shù)據(jù)；

所述數(shù)據(jù)庫(kù)用于接收并存儲(chǔ)所述全量采集同步請(qǐng)求、所述審核數(shù)據(jù)；

所述采集同步模塊依據(jù)所述審核數(shù)據(jù)進(jìn)行采集數(shù)據(jù)的同步；

所述采集同步模塊依據(jù)所述審核數(shù)據(jù)進(jìn)行采集同步的具體步驟如下：

步驟A1：采集同步模塊接收審核數(shù)據(jù)，審核數(shù)據(jù)內(nèi)存儲(chǔ)有采集數(shù)據(jù)，采集同步模塊判斷采集數(shù)據(jù)是否指定了分片列；

步驟A2：若沒有指定分片列，則判斷采集數(shù)據(jù)中是否包含PK字段、UK字段；

若包含PK字段、UK字段則按照PK字段、UK字段的優(yōu)先級(jí)進(jìn)行分片；

若未包含PK字段、UK字段則不進(jìn)行分片，所有采集數(shù)據(jù)作為一個(gè)分片進(jìn)行采集同步；

若指定分片列，則判斷該分片列是否包含索引；

步驟A3：若包含索引，則依據(jù)索引進(jìn)行分片采集同步；

若不包含索引，則不進(jìn)行分片，所有采集數(shù)據(jù)作為一個(gè)分片進(jìn)行采集同步；

步驟A4：對(duì)采集數(shù)據(jù)按照步驟A2-A3的分類進(jìn)行采集同步。

2.如權(quán)利要求1所述的一種數(shù)據(jù)采集同步系統(tǒng)，其特征在于，所述采集同步模塊包含全量采集同步模塊、日志采集模塊、流量式增量采集轉(zhuǎn)換模塊、流式采集清洗模塊、可視化管理配置模塊；

所述全量采集同步模塊依據(jù)所述審核數(shù)據(jù)進(jìn)行采集同步，所述全量采集同步模塊還用于通過Flink內(nèi)存計(jì)算框架對(duì)采集數(shù)據(jù)進(jìn)行全量數(shù)據(jù)的采集同步；

所述日志采集模塊用于與所述數(shù)據(jù)庫(kù)創(chuàng)建連接監(jiān)控所述數(shù)據(jù)庫(kù)內(nèi)的binlog日志數(shù)據(jù)并同步投遞到kafka中進(jìn)行保存；

所述流量式增量采集轉(zhuǎn)換模塊用于從所述kafka中獲取所述binlog日志數(shù)據(jù)并進(jìn)行流式處理轉(zhuǎn)換；

所述流式采集清洗模塊用于從所述kafka中獲取所述binlog日志數(shù)據(jù)并進(jìn)行清洗；

所述可視化管理配置模塊用于進(jìn)行采集同步的開始、停止、查詢同步狀態(tài)、狀態(tài)異常告警、數(shù)據(jù)流實(shí)時(shí)規(guī)則配置、數(shù)據(jù)流實(shí)時(shí)在線清洗、數(shù)據(jù)流實(shí)時(shí)結(jié)構(gòu)化、數(shù)據(jù)量化；

所述流量式增量采集轉(zhuǎn)換模塊對(duì)所述binlog日志數(shù)據(jù)進(jìn)行流式處理轉(zhuǎn)換的具體步驟如下：

步驟B1：由用戶端發(fā)出增量采集同步請(qǐng)求至采集同步模塊；

步驟B2：采集同步模塊先判斷是否有全量采集同步請(qǐng)求任務(wù)在運(yùn)行；

若有則暫停增量采集同步請(qǐng)求任務(wù)，一段時(shí)間后再次判斷是否有全量采集同步請(qǐng)求任務(wù)在運(yùn)行，直至全量采集同步請(qǐng)求任務(wù)完成；

若沒有則判斷增量采集同步請(qǐng)求是否有分發(fā)規(guī)則；

步驟B3：若有分發(fā)規(guī)則則按照該分發(fā)規(guī)則將binlog日志數(shù)據(jù)分發(fā)到不同的topic中；

若沒有分發(fā)規(guī)則則將binlog日志數(shù)據(jù)分發(fā)到固定的topic中；

步驟B4：將topic中的binlog日志數(shù)據(jù)與采集數(shù)據(jù)合并得到全量數(shù)據(jù)。

3.如權(quán)利要求2所述的一種數(shù)據(jù)采集同步系統(tǒng)，其特征在于，所述日志采集模塊還用于與第三方日志采集組件連接獲取日志數(shù)據(jù)進(jìn)行同步，具體包含以下步驟；

步驟C1：日志采集模塊獲取第三方日志采集組件傳輸?shù)娜罩緮?shù)據(jù)；

步驟C2：判斷該日志數(shù)據(jù)是否配置清洗規(guī)則；

若沒有配置清洗規(guī)則，則將日志數(shù)據(jù)加入未知數(shù)據(jù)中；

若配置了清洗規(guī)則，則按照清洗規(guī)則進(jìn)行清洗并進(jìn)行結(jié)構(gòu)化處理生成結(jié)構(gòu)化日志數(shù)據(jù)；

步驟C3：將未知數(shù)據(jù)與結(jié)構(gòu)化日志數(shù)據(jù)整合并傳輸至數(shù)據(jù)庫(kù)中。

4.如權(quán)利要求3所述的一種數(shù)據(jù)采集同步系統(tǒng)，其特征在于，所述第三方日志采集組件為Flume日志采集組件或Filebaet日志采集組件。