[發(fā)明專利]數(shù)據(jù)處理方法及裝置

在審

• 申請?zhí)枺?/td>: 202210445104.8
• 申請日: 2022-04-26
• 公開（公告）號: CN114780529A
• 公開（公告）日: 2022-07-22
• 發(fā)明（設(shè)計(jì)）人: 顧超;劉立力
• 申請（專利權(quán)）人: 湖南快樂陽光互動娛樂傳媒有限公司
• 主分類號: G06F16/215
• 分類號: G06F16/215;G06F16/22;G06F16/2455;G06F11/30
• 代理公司: 北京集佳知識產(chǎn)權(quán)代理有限公司 11227
• 代理人: 李偉

本發(fā)明提供了一種數(shù)據(jù)處理方法及裝置，該方法包括：接收數(shù)據(jù)清洗任務(wù)；獲取數(shù)據(jù)清洗任務(wù)對應(yīng)的數(shù)據(jù)來源信息和數(shù)據(jù)輸出信息；獲取FLinkSQL腳本，并應(yīng)用FLinkSQL腳本對數(shù)據(jù)來源信息對應(yīng)的第一數(shù)據(jù)流和數(shù)據(jù)輸出信息對應(yīng)的第二數(shù)據(jù)流進(jìn)行數(shù)據(jù)清洗；啟用任務(wù)監(jiān)控模塊對數(shù)據(jù)清洗過程進(jìn)行實(shí)時監(jiān)控以確定是否存在任務(wù)積壓；當(dāng)不存在任務(wù)積壓，獲取清洗結(jié)果，判斷清洗結(jié)果是否滿足清洗條件；當(dāng)滿足清洗條件，將清洗結(jié)果發(fā)送至Flink集群，通過Flink集群存儲并輸出清洗結(jié)果。應(yīng)用本發(fā)明提供的方法，可以應(yīng)用FlinkSQL腳本實(shí)現(xiàn)自動化清洗，并且對清洗過程進(jìn)行監(jiān)控，減少人工參與，提高數(shù)據(jù)處理的效率和用戶體驗(yàn)。

技術(shù)領(lǐng)域

本發(fā)明涉及數(shù)據(jù)處理技術(shù)領(lǐng)域，特別是涉及一種數(shù)據(jù)處理方法及裝置。

背景技術(shù)

實(shí)時數(shù)據(jù)清洗是整個數(shù)據(jù)分析過程中一個不可缺少的環(huán)節(jié)，其清洗結(jié)果質(zhì)量直接關(guān)系到數(shù)據(jù)模型的效果和最終用于決策的效果，而數(shù)據(jù)清洗的過程往往占整個數(shù)據(jù)分析的50％甚至更多的時間。

現(xiàn)有技術(shù)中，對數(shù)據(jù)進(jìn)行清洗的過程涉及到實(shí)時的數(shù)據(jù)流清洗，而對于數(shù)據(jù)流的清洗需要注意數(shù)據(jù)清洗過程的效率以及準(zhǔn)確性，但現(xiàn)有技術(shù)中對數(shù)據(jù)流的實(shí)時清洗需要用戶參與整個操作，導(dǎo)致數(shù)據(jù)清洗的效率較低，且影響用戶的操作體驗(yàn)。

發(fā)明內(nèi)容

有鑒于此，本發(fā)明提供一種數(shù)據(jù)處理方法，通過該方法，可以應(yīng)用Flink SQL腳本實(shí)現(xiàn)自動化清洗，并且對清洗過程進(jìn)行監(jiān)控，減少人工參與，提高數(shù)據(jù)處理的效率和用戶體驗(yàn)。

本發(fā)明還提供了一種數(shù)據(jù)處理裝置，用以保證上述方法在實(shí)際中的實(shí)現(xiàn)及應(yīng)用。

一種數(shù)據(jù)處理方法，包括：

接收用戶經(jīng)由數(shù)據(jù)中心發(fā)出的數(shù)據(jù)清洗任務(wù)；

獲取所述數(shù)據(jù)清洗任務(wù)對應(yīng)的數(shù)據(jù)來源信息和數(shù)據(jù)輸出信息，所述數(shù)據(jù)來源信息及數(shù)據(jù)輸出信息為數(shù)據(jù)的存儲信息，所述數(shù)據(jù)的存儲信息為數(shù)據(jù)庫的數(shù)據(jù)表信息或kafka的主題信息；

基于預(yù)設(shè)的表創(chuàng)建語法，對所述數(shù)據(jù)來源信息對應(yīng)的數(shù)據(jù)流及所述數(shù)據(jù)輸出信息對應(yīng)的數(shù)據(jù)流進(jìn)行處理，獲得所述數(shù)據(jù)來源信息對應(yīng)的第一數(shù)據(jù)流和所述數(shù)據(jù)輸出信息對應(yīng)的第二數(shù)據(jù)流；

獲取所述用戶預(yù)先提交的數(shù)據(jù)清洗FLink SQL腳本，并應(yīng)用所述FLink SQL腳本對所述第一數(shù)據(jù)流，和所述第二數(shù)據(jù)流進(jìn)行數(shù)據(jù)清洗；

啟用預(yù)設(shè)的任務(wù)監(jiān)控模塊對數(shù)據(jù)清洗過程進(jìn)行實(shí)時監(jiān)控，確定所述數(shù)據(jù)清洗任務(wù)是否存在任務(wù)積壓；

當(dāng)所述數(shù)據(jù)清洗任務(wù)不存在任務(wù)積壓時，獲取數(shù)據(jù)清洗后的清洗結(jié)果，并確定所述清洗結(jié)果是否滿足預(yù)設(shè)的清洗條件；

當(dāng)所述清洗結(jié)果滿足所述清洗條件時，將所述清洗結(jié)果發(fā)送至預(yù)先設(shè)置的Flink集群，并通過所述Flink集群存儲并輸出所述清洗結(jié)果。

上述的方法，可選的，所述獲取所述數(shù)據(jù)清洗任務(wù)對應(yīng)的數(shù)據(jù)來源信息和數(shù)據(jù)輸出信息，包括：

連接所述數(shù)據(jù)中心的數(shù)據(jù)庫；

獲取所述數(shù)據(jù)庫中所述數(shù)據(jù)清洗任務(wù)對應(yīng)的任務(wù)標(biāo)識碼；

基于所述任務(wù)標(biāo)識碼提取所述數(shù)據(jù)庫中所述數(shù)據(jù)清洗任務(wù)對應(yīng)的任務(wù)包；

解析所述任務(wù)包，獲取所述任務(wù)包中所述數(shù)據(jù)清洗任務(wù)對應(yīng)的數(shù)據(jù)來源信息和數(shù)據(jù)輸出信息。

上述的方法，可選的，所述啟用預(yù)設(shè)的任務(wù)監(jiān)控模塊對數(shù)據(jù)清洗過程進(jìn)行實(shí)時監(jiān)控，確定所述數(shù)據(jù)清洗任務(wù)是否存在任務(wù)積壓，包括：

啟動預(yù)先設(shè)置的Kafka消費(fèi)組水位邏輯模塊；

應(yīng)用所述Kafka消費(fèi)組水位邏輯模塊，實(shí)時記錄所述FLink SQL腳本每次消費(fèi)所述第一數(shù)據(jù)流和所述第二數(shù)據(jù)流進(jìn)行數(shù)據(jù)清洗的消費(fèi)延遲信息；