[發(fā)明專利]實(shí)時(shí)數(shù)據(jù)流處理方法、裝置、設(shè)備、及可讀存儲(chǔ)介質(zhì)

在審

• 申請(qǐng)?zhí)枺?/td>: 202011024249.8
• 申請(qǐng)日: 2020-09-25
• 公開（公告）號(hào): CN112416972A
• 公開（公告）日: 2021-02-26
• 發(fā)明（設(shè)計(jì)）人: 陳健;蔡雪峰
• 申請(qǐng)（專利權(quán)）人: 上海嗶哩嗶哩科技有限公司
• 主分類號(hào): G06F16/2455
• 分類號(hào): G06F16/2455;G06F16/28
• 代理公司: 北京英特普羅知識(shí)產(chǎn)權(quán)代理有限公司 11015
• 代理人: 程超

1.一種實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述方法包括：

獲取多個(gè)數(shù)據(jù)流在預(yù)設(shè)時(shí)間窗口內(nèi)產(chǎn)生的數(shù)據(jù)項(xiàng)；其中，每個(gè)數(shù)據(jù)項(xiàng)包括：一個(gè)索引字段；

將所述多個(gè)數(shù)據(jù)流中的包含同一索引字段的數(shù)據(jù)項(xiàng)進(jìn)行合并，并將合并后的合并數(shù)據(jù)項(xiàng)存儲(chǔ)到緩存中；

當(dāng)所述合并數(shù)據(jù)項(xiàng)在所述緩存中的存儲(chǔ)時(shí)長達(dá)到預(yù)設(shè)閾值時(shí)，將所述合并數(shù)據(jù)項(xiàng)存儲(chǔ)到目標(biāo)數(shù)據(jù)庫中。

2.根據(jù)權(quán)利要求1所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，在所述將合并后的合并數(shù)據(jù)項(xiàng)存儲(chǔ)到緩存中之后，且所述合并數(shù)據(jù)項(xiàng)在所述緩存中的存儲(chǔ)時(shí)長未達(dá)到所述預(yù)設(shè)閾值時(shí)，所述方法還包括：

當(dāng)任一數(shù)據(jù)流中有撤回流操作時(shí)，獲取所述數(shù)據(jù)流中的更新數(shù)據(jù)項(xiàng)；

根據(jù)所述更新數(shù)據(jù)項(xiàng)中的索引字段，從所述緩存中查找到包含所述索引字段的合并數(shù)據(jù)項(xiàng)，并根據(jù)所述更新數(shù)據(jù)項(xiàng)對(duì)所述合并數(shù)據(jù)項(xiàng)進(jìn)行更新。

3.根據(jù)權(quán)利要求2所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述方法還包括：

獲取預(yù)設(shè)的合并配置文件；其中，所述合并配置文件包括：多個(gè)目標(biāo)索引字段、以及每個(gè)目標(biāo)索引字段所對(duì)應(yīng)的合并規(guī)則。

4.根據(jù)權(quán)利要求3所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述將所述多個(gè)數(shù)據(jù)流中的包含同一索引字段的數(shù)據(jù)項(xiàng)進(jìn)行合并，并將合并后的合并數(shù)據(jù)項(xiàng)存儲(chǔ)到緩存中，包括：

針對(duì)所述合并配置文件中的一個(gè)目標(biāo)索引字段，判斷在所述多個(gè)數(shù)據(jù)流中是否存在包含所述目標(biāo)索引字段的數(shù)據(jù)項(xiàng)；

若是，則將所有包含所述目標(biāo)索引字段的數(shù)據(jù)項(xiàng)發(fā)送至與所述目標(biāo)索引字段對(duì)應(yīng)的合并節(jié)點(diǎn)上；

通過所述合并節(jié)點(diǎn)，按照所述合并配置文件中的與所述目標(biāo)索引字段對(duì)應(yīng)的合并規(guī)則，將所有包含所述目標(biāo)索引字段的數(shù)據(jù)項(xiàng)合并為合并數(shù)據(jù)項(xiàng)；

將所述合并數(shù)據(jù)項(xiàng)、以及所有包含所述目標(biāo)索引字段的數(shù)據(jù)項(xiàng)存儲(chǔ)到所述合并節(jié)點(diǎn)上的緩存中。

5.根據(jù)權(quán)利要求4所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述根據(jù)所述更新數(shù)據(jù)項(xiàng)中的索引字段，從所述緩存中查找到包含所述索引字段的合并數(shù)據(jù)項(xiàng)，并根據(jù)所述更新數(shù)據(jù)項(xiàng)對(duì)所述合并數(shù)據(jù)項(xiàng)進(jìn)行更新，包括：

將所述更新數(shù)據(jù)項(xiàng)存儲(chǔ)到與所述索引字段對(duì)應(yīng)的合并節(jié)點(diǎn)上的緩存中；

確定出所述更新數(shù)據(jù)項(xiàng)的數(shù)據(jù)流類型；

從所述緩存中刪除包含所述索引字段、且與所述數(shù)據(jù)流類型一致的數(shù)據(jù)項(xiàng)，以及從所述緩存中刪除包含所述索引字段的合并數(shù)據(jù)項(xiàng)；

通過所述合并節(jié)點(diǎn)，按照所述合并配置文件中的與所述索引字段對(duì)應(yīng)的合并規(guī)則，將所述緩存中所有包含所述索引字段的數(shù)據(jù)項(xiàng)合并為新的合并數(shù)據(jù)項(xiàng)；

將所述新的合并數(shù)據(jù)項(xiàng)存儲(chǔ)到所述緩存中。

6.根據(jù)權(quán)利要求5所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述確定出所述更新數(shù)據(jù)項(xiàng)的數(shù)據(jù)流類型，包括：

根據(jù)所述更新數(shù)據(jù)項(xiàng)中包含的字段長度確定出所述更新數(shù)據(jù)項(xiàng)的數(shù)據(jù)流類型。

7.根據(jù)權(quán)利要求3所述的實(shí)時(shí)數(shù)據(jù)流處理方法，其特征在于，所述方法還包括：

若任一數(shù)據(jù)項(xiàng)的索引字段不包含在所述合并配置文件中，則將所述數(shù)據(jù)項(xiàng)存儲(chǔ)到所述緩存中；

當(dāng)所述數(shù)據(jù)項(xiàng)在所述緩存中的存儲(chǔ)時(shí)長達(dá)到所述預(yù)設(shè)閾值時(shí)，將所述數(shù)據(jù)項(xiàng)存儲(chǔ)到所述目標(biāo)數(shù)據(jù)庫中。

8.一種實(shí)時(shí)數(shù)據(jù)流處理裝置，其特征在于，所述裝置包括：

獲取模塊，用于獲取多個(gè)數(shù)據(jù)流在預(yù)設(shè)時(shí)間窗口內(nèi)產(chǎn)生的數(shù)據(jù)項(xiàng)；其中，每個(gè)數(shù)據(jù)項(xiàng)包括：一個(gè)索引字段；

合并模塊，用于將所述多個(gè)數(shù)據(jù)流中的包含同一索引字段的數(shù)據(jù)項(xiàng)進(jìn)行合并，并將合并后的合并數(shù)據(jù)項(xiàng)存儲(chǔ)到緩存中；

存儲(chǔ)模塊，用于當(dāng)所述合并數(shù)據(jù)項(xiàng)在所述緩存中的存儲(chǔ)時(shí)長達(dá)到預(yù)設(shè)閾值時(shí)，將所述合并數(shù)據(jù)項(xiàng)存儲(chǔ)到目標(biāo)數(shù)據(jù)庫中。

9.一種計(jì)算機(jī)設(shè)備，所述計(jì)算機(jī)設(shè)備包括：存儲(chǔ)器、處理器以及存儲(chǔ)在所述存儲(chǔ)器上并可在所述處理器上運(yùn)行的計(jì)算機(jī)程序，其特征在于，所述處理器執(zhí)行所述計(jì)算機(jī)程序時(shí)實(shí)現(xiàn)權(quán)利要求1至7中任一項(xiàng)所述方法的步驟。

10.一種計(jì)算機(jī)可讀存儲(chǔ)介質(zhì)，其上存儲(chǔ)有計(jì)算機(jī)程序，其特征在于，所述計(jì)算機(jī)程序被處理器執(zhí)行時(shí)實(shí)現(xiàn)權(quán)利要求1至7中任一項(xiàng)所述方法的步驟。