[發明專利]基于分布式數據流處理引擎的數據庫導數方法、設備和介質

有效

• 申請號: 202110254713.0
• 申請日: 2021-03-05
• 公開（公告）號: CN112905668B
• 公開（公告）日: 2023-06-06
• 發明（設計）人: 張靈星;王海霖;陳黃;張國慶
• 申請（專利權）人: 北京中經惠眾科技有限公司
• 主分類號: G06F16/2458
• 分類號: G06F16/2458;G06F16/2457;G06F16/27
• 代理公司: 北京市漢坤律師事務所 11602
• 代理人: 初媛媛;吳麗麗

1.一種基于分布式數據流處理引擎的數據庫導數方法，包括：

從消息系統的多個分區讀取待導入數據庫的多個數據；

將所述多個數據分別存儲至所述分布式數據流處理引擎的存儲單元中；

將所述存儲單元中的數據導入數據庫；

根據預設規則觸發所述分布式數據流處理引擎執行檢查點設置操作，

其中，所述檢查點設置操作包括：

響應于所述分布式數據流處理引擎被觸發執行檢查點設置操作，獲取當前從多個分區中的每一分區所讀取的數據的位置參數，以使得當前從多個分區中的每一分區所讀取的數據能夠作為檢查數據；

存儲多個檢查數據的位置參數，以使得基于所存儲的所述多個檢查數據的位置參數能夠實現數據庫導數的斷點續傳；

為所述多個檢查數據中的每一個檢查數據標注屏障標記；以及

響應于被標注屏障標記的多個檢查數據均被成功讀入，確定完成檢查點設置操作。

2.根據權利要求1所述的方法，其中，根據預設規則觸發所述分布式數據流處理引擎執行檢查點設置操作包括：

按照預設周期觸發所述分布式數據流處理引擎執行檢查點設置操作。

3.根據權利要求2所述的方法，其中，按照預設周期觸發所述分布式數據流處理引擎執行檢查點設置操作包括：

在每一周期中，對從消息系統讀取的數據進行計數，并且對從消息系統讀取數據的持續時長進行計時；

響應于從消息系統讀取的數據的計數達到預設數量或者從消息系統讀取數據的持續時長達到預設時長，觸發所述分布式數據流處理引擎執行檢查點設置操作。

4.根據權利要求1所述的方法，其中，所述存儲單元包括所述多個檢查數據和位置參數之間的對應關系，

所述方法還包括：

響應于檢測到數據庫的導數過程出現中斷，基于多個檢查數據的位置參數，繼續執行將存儲單元中的數據導入數據庫。

5.根據權利要求1所述的方法，其中，每一檢查數據的位置參數包括該檢查數據所對應的與分區相關的第一編碼，以及與該檢查數據的讀取順序相關的第二編碼。

6.根據權利要求5所述的方法，其中，針對從每一個分區讀取的多個數據，累計所讀取的數據的個數，

其中，針對該分區，將所確定的檢查數據對應的所述累計得到的個數作為該檢查數據的所述第二編碼。

7.根據權利要求1所述的方法，其中，存儲多個檢查數據的位置參數包括：

將之前存儲的多個檢查數據的位置參數更新為當前所確定的多個檢查數據的位置參數。

8.根據權利要求7所述的方法，還包括：

累計執行檢查點設置操作的次數。

9.根據權利要求1所述的方法，其中，響應于從消息系統的多個分區讀取的數據被成功讀入，繼續從該數據對應的分區讀取數據。

10.根據權利要求1所述的方法，其中，響應于被標注屏障標記的某一檢查數據被成功讀入，繼續從該檢查數據對應的分區讀取數據。

11.根據權利要求1-10中任一項所述的方法，其中，所述分布式數據流處理引擎包括連接器，所述分布式數據流處理引擎通過連接器從消息系統的多個分區讀取待導入數據庫的多個數據。

12.根據權利要求11所述的方法，其中，所述分布式數據流處理引擎為flink。

13.根據權利要求1-10中任一項所述的方法，其中，所述消息系統為kafka消息系統。