[發(fā)明專利]基于任務(wù)并行處理數(shù)據(jù)流的方法、裝置、設(shè)備及存儲(chǔ)介質(zhì)

有效

• 申請(qǐng)?zhí)枺?/td>: 201711381582.2
• 申請(qǐng)日: 2017-12-20
• 公開（公告）號(hào): CN108121792B
• 公開（公告）日: 2020-06-26
• 發(fā)明（設(shè)計(jì)）人: 楊強(qiáng);陳雨強(qiáng);戴文淵;焦英翔;石光川
• 申請(qǐng)（專利權(quán)）人: 第四范式（北京）技術(shù)有限公司
• 主分類號(hào): G06F16/2453
• 分類號(hào): G06F16/2453;G06F16/2455;G06F9/50
• 代理公司: 北京銘碩知識(shí)產(chǎn)權(quán)代理有限公司 11286
• 代理人: 蘇銀虹;徐璐璐

本發(fā)明公開了一種數(shù)據(jù)流并行處理的方法、裝置、設(shè)備及存儲(chǔ)介質(zhì)。通過確定的多個(gè)工作線程之中的每一個(gè)工作線程，分別從任務(wù)隊(duì)列中取出待處理任務(wù)，以針對(duì)取出的待處理任務(wù)進(jìn)行處理，其中，待處理任務(wù)由數(shù)據(jù)流之中的待操作分批數(shù)據(jù)與數(shù)據(jù)流處理之中的相應(yīng)操作步驟打包而成。本發(fā)明公開的這種基于任務(wù)生成式的并行化運(yùn)行機(jī)制，對(duì)于基于不同操作步驟打包生成的任務(wù)，在實(shí)際運(yùn)行過程中會(huì)根據(jù)其耗時(shí)自動(dòng)地調(diào)整并行度。

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

本發(fā)明涉及數(shù)據(jù)科學(xué)領(lǐng)域，特別是涉及一種基于任務(wù)對(duì)數(shù)據(jù)流并行處理的方法、裝置、設(shè)備及存儲(chǔ)介質(zhì)。

背景技術(shù)

在數(shù)據(jù)處理業(yè)務(wù)涉及的數(shù)據(jù)量很大的情況下，通常需要利用多線程并行執(zhí)行來使業(yè)務(wù)的總體執(zhí)行時(shí)間減小。線程是操作系統(tǒng)進(jìn)行運(yùn)算調(diào)度的單位，計(jì)算必須以線程的方式交給操作系統(tǒng)執(zhí)行，在實(shí)際計(jì)算時(shí)，線程會(huì)被分配給某個(gè)物理核來進(jìn)行計(jì)算。盡量多的使用機(jī)器的物理資源可以降低任務(wù)的總時(shí)間開銷。

一種利用多線程的思路是將需要處理的數(shù)據(jù)分成多批，每個(gè)線程負(fù)責(zé)處理一批數(shù)據(jù)。這種多線程執(zhí)行方案的問題是，數(shù)據(jù)的處理一般包括讀寫和計(jì)算，數(shù)據(jù)的讀寫是硬盤操作，而計(jì)算是諸如CPU操作的處理，一個(gè)線程如果在多種硬件設(shè)備上切換操作，會(huì)帶來很大的額外開銷。另外在當(dāng)前線程等待數(shù)據(jù)讀入時(shí)，其對(duì)應(yīng)的CPU并沒有執(zhí)行操作，相當(dāng)于浪費(fèi)了同一時(shí)間的CPU資源。

現(xiàn)有技術(shù)主要靠異步IO解決這個(gè)問題。假設(shè)數(shù)據(jù)處理過程包括數(shù)據(jù)讀取、數(shù)據(jù)計(jì)算、計(jì)算結(jié)果存儲(chǔ)三個(gè)步驟，則可以設(shè)置三個(gè)線程，線程一只負(fù)責(zé)從硬盤讀取數(shù)據(jù)，并將數(shù)據(jù)放入緩沖區(qū)一，線程二只負(fù)責(zé)從緩沖區(qū)一讀取數(shù)據(jù)，對(duì)其進(jìn)行計(jì)算并將計(jì)算結(jié)果寫入緩沖區(qū)二，線程三只負(fù)責(zé)將緩沖區(qū)二的數(shù)據(jù)寫入新文件(這里一個(gè)線程在完成對(duì)全部數(shù)據(jù)的操作之前不會(huì)停止)。一般來說，為了防止內(nèi)存中的數(shù)據(jù)過大，緩沖區(qū)都設(shè)有可緩存的上限數(shù)量。

假設(shè)數(shù)據(jù)的讀寫對(duì)于單位數(shù)據(jù)所消耗的時(shí)間為一個(gè)時(shí)間單位，計(jì)算操作所消耗的時(shí)間較多，為三個(gè)時(shí)間單位，則可以通過增加計(jì)算操作的線程數(shù)來抵消上述運(yùn)算速度的差異。如可以設(shè)置五個(gè)線程，兩個(gè)分別負(fù)責(zé)讀寫，三個(gè)負(fù)責(zé)計(jì)算，則緩沖區(qū)內(nèi)的數(shù)據(jù)不會(huì)不斷累加，計(jì)算資源在整個(gè)任務(wù)過程中都處于被使用的狀態(tài)。

然而異步IO的問題是，當(dāng)數(shù)據(jù)的處理變得復(fù)雜時(shí)，如假設(shè)數(shù)據(jù)的處理過程包括多個(gè)操作步驟，且每個(gè)操作步驟的時(shí)間資源消耗都不相同，這種情況下就很難將線程比例配置到最合理。而且在異步IO的整個(gè)計(jì)算過程中，開啟的線程數(shù)是所有操作步驟所需線程數(shù)之和。由于計(jì)算機(jī)上的物理CPU個(gè)數(shù)是固定的，過多的線程會(huì)增加系統(tǒng)調(diào)度時(shí)的切換開銷。而相加的關(guān)系導(dǎo)致總線程數(shù)與數(shù)據(jù)的具體處理流程有關(guān)，通用計(jì)算框架不得不考慮各種數(shù)據(jù)處理業(yè)務(wù)的具體特性來對(duì)線程數(shù)進(jìn)行控制。

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

本發(fā)明的一個(gè)目的在于提供一種基于任務(wù)生成式的并行化處理方案，其能夠根據(jù)數(shù)據(jù)處理流程的復(fù)雜度自適應(yīng)調(diào)整并行度。

根據(jù)本發(fā)明的一個(gè)方面，提供了一種基于任務(wù)對(duì)數(shù)據(jù)流并行處理的方法，包括：確定多個(gè)工作線程；通過多個(gè)工作線程之中的每一個(gè)工作線程，分別從任務(wù)隊(duì)列中取出待處理任務(wù)，以針對(duì)取出的待處理任務(wù)進(jìn)行處理，其中，待處理任務(wù)由數(shù)據(jù)流之中的待操作分批數(shù)據(jù)與數(shù)據(jù)流處理之中的相應(yīng)操作步驟打包而成。

可選地，該方法還可以包括：每一個(gè)工作線程將處理后得到的操作結(jié)果數(shù)據(jù)作為新的待操作分批數(shù)據(jù)與相應(yīng)操作步驟的后續(xù)操作步驟打包成新的待處理任務(wù)，以放入任務(wù)隊(duì)列。

可選地，每一個(gè)工作線程將處理后得到的操作結(jié)果數(shù)據(jù)作為新的待操作分批數(shù)據(jù)與其分支條件被滿足的分支的后續(xù)操作步驟打包成新的待處理任務(wù)，以放入任務(wù)隊(duì)列。

可選地，該方法還可以包括：確定單獨(dú)的源線程；通過源線程來專門循環(huán)地將數(shù)據(jù)流之中的待操作分批數(shù)據(jù)與數(shù)據(jù)流處理之中的起始操作步驟打包成待處理任務(wù)，以放入任務(wù)隊(duì)列。

可選地，該方法還可以包括：監(jiān)控對(duì)數(shù)據(jù)流并行處理過程中的內(nèi)存使用情況；在當(dāng)前使用的內(nèi)存超過預(yù)定閾值的情況下，源線程暫停向任務(wù)隊(duì)列中放入新的待處理任務(wù)。