[發(fā)明專(zhuān)利]一種流式數(shù)據(jù)處理自適應(yīng)速率控制方法

有效

• 申請(qǐng)?zhí)枺?/td>: 201710011464.6
• 申請(qǐng)日: 2017-01-09
• 公開(kāi)（公告）號(hào): CN106648904B
• 公開(kāi)（公告）日: 2020-06-12
• 發(fā)明（設(shè)計(jì)）人: 申彥明;李曉東
• 申請(qǐng)（專(zhuān)利權(quán)）人: 大連理工大學(xué)
• 主分類(lèi)號(hào): G06F9/50
• 分類(lèi)號(hào): G06F9/50;G06F16/2455;G06F16/182
• 代理公司: 大連理工大學(xué)專(zhuān)利中心 21200
• 代理人: 梅洪玉

1.一種流式數(shù)據(jù)處理自適應(yīng)速率控制方法，其特征在于，包括如下步驟：

第一步，數(shù)據(jù)接收和分區(qū)緩存

流式數(shù)據(jù)處理系統(tǒng)接收的數(shù)據(jù)，將一條完整的數(shù)據(jù)視為一行，或者使用無(wú)歧義的特殊分隔符對(duì)數(shù)據(jù)進(jìn)行分割；使用消息隊(duì)列作為數(shù)據(jù)源對(duì)數(shù)據(jù)進(jìn)行匯總，并對(duì)數(shù)據(jù)按照分區(qū)規(guī)則將數(shù)據(jù)保存到消息隊(duì)列中的某個(gè)分區(qū)中；

第二步，數(shù)據(jù)預(yù)分片

從消息隊(duì)列中讀取數(shù)據(jù)前，將消息隊(duì)列中單個(gè)數(shù)據(jù)分片映射到多個(gè)流式數(shù)據(jù)處理集群的分片中，映射的數(shù)量根據(jù)集群虛擬核心數(shù)量進(jìn)行設(shè)置；

第三步，處理速率實(shí)時(shí)反饋控制

從消息隊(duì)列中讀取數(shù)據(jù)，并根據(jù)當(dāng)前的數(shù)據(jù)處理速率上限確定最多可以從消息隊(duì)列中讀取的數(shù)據(jù)數(shù)量，避免讀取的數(shù)據(jù)數(shù)量超過(guò)計(jì)算集群的最大處理能力；實(shí)時(shí)速率控制器的處理速率上限計(jì)算方法如下：

3.1)通過(guò)慢啟動(dòng)的方式開(kāi)始進(jìn)行調(diào)整，初始的數(shù)據(jù)處理速率上限設(shè)置為消息隊(duì)列的每個(gè)分片每秒獲讀取n條數(shù)據(jù)，50n1000；

3.2)第一批完成后，對(duì)以后每一批按照以下的步驟對(duì)數(shù)據(jù)處理速率上限進(jìn)行計(jì)算；

3.2.1)在批次開(kāi)始時(shí)，提交當(dāng)前批次的開(kāi)始時(shí)間t_batchst給速率控制器；

3.2.2)在批次完成時(shí)，提交當(dāng)前批次的處理結(jié)束時(shí)間t_batched，處理該批次用時(shí)t_proc_last，該批次在批次隊(duì)列中的等待時(shí)間t_wait_last，該批次處理數(shù)據(jù)的數(shù)量elemnum_last；

3.2.3)在當(dāng)前批次提交時(shí)，對(duì)當(dāng)前批次的數(shù)據(jù)處理速率上限進(jìn)行計(jì)算，分為以下三種情況：第一種情況，在當(dāng)前批次提交時(shí)，上一批次執(zhí)行完成，且上一批次數(shù)據(jù)的處理時(shí)間與流式數(shù)據(jù)批處理集群的批次間隔t_interval相差大于松弛參數(shù)p_relx；第二種情況，當(dāng)前批次提交時(shí)，上一批次執(zhí)行完成，且上一批次數(shù)據(jù)的處理時(shí)間位于t_interval-p_relx和t_interval之間；p_relx＝min(50，t_interval×0.05)；第三種情況，當(dāng)前批次提交時(shí)，上一批次尚未完成；

第一種情況和第三種情況都基于PID控制算法對(duì)數(shù)據(jù)處理速率上限計(jì)算：

newRate＝latestRate-K_p×error-K_i×historicalError-K_d×dError (1)

其中，latestRate為最新處理完成批次的數(shù)據(jù)處理速率上限；K_p為比例系數(shù)，取值為1；error為數(shù)據(jù)處理速率的估計(jì)誤差；K_i為積分系數(shù)，取值為0.2；historicalError為數(shù)據(jù)處理速率的累積誤差；K_d為微分系數(shù)，為了減小噪聲數(shù)據(jù)對(duì)系統(tǒng)的影響，取值為0；dError為偏差速率變化率；

數(shù)據(jù)處理速率的估計(jì)誤差error為：

其中，elemnum_last為最新處理完成的批次處理數(shù)據(jù)的數(shù)量；t_proc_last為最新處理完成的批次處理的時(shí)間；K_block為阻塞系數(shù)，阻塞系數(shù)取值0.3；blockTime為估計(jì)的處理時(shí)間增加值；當(dāng)為第一種情況時(shí)，blockTime取值為0，此時(shí)由公式(2)計(jì)算得到的結(jié)果為數(shù)據(jù)處理速率的估計(jì)誤差；當(dāng)為第三種情況時(shí)，blockTime計(jì)算如公式(3)所示：

blockTime＝max(t_interval-(submit_time-t_batchst),b_relx) (3)

其中，t_interval為批次提交的間隔；submit_time為任務(wù)的提交時(shí)間；在批次間隔大于1000ms時(shí)，b_relx＝50ms，當(dāng)批次間隔小于1000ms時(shí)，b_relx＝t_interval×0.05；

數(shù)據(jù)處理速率的累積誤差historicalError的計(jì)算如公式(4)所示：

其中，t_wait_last為最新處理完成的批次在批次隊(duì)列中等待的時(shí)間；當(dāng)為第一種情況時(shí)，blockTime取值為0；當(dāng)為第三種情況時(shí)，blockTime計(jì)算如公式(3)所示；processingRate表示最新處理完成批次的數(shù)據(jù)處理速率，其計(jì)算如公式(5)所示：

偏差速率變化率dError的計(jì)算如公式(6)所示：

其中，latestError表示前一個(gè)處理完成批次的速率估計(jì)誤差，delaySinceUpdate定義為前一個(gè)處理完成批次和最近完成批次之間經(jīng)過(guò)的時(shí)間，其計(jì)算方式如公式(7)所示：

delaySinceUpdate＝lastProcessingEndTime-latestTime (7)

第二種情況，使用最新處理完成批次的數(shù)據(jù)處理速率上限latestRate作為當(dāng)前批次的數(shù)據(jù)處理速率上限newRate；

第四步，數(shù)據(jù)的讀取和實(shí)時(shí)處理；流式數(shù)據(jù)批處理集群按照第三步計(jì)算的數(shù)據(jù)處理速率上限和第二步預(yù)分片規(guī)則對(duì)數(shù)據(jù)進(jìn)行讀取；在本批次數(shù)據(jù)處理完成之后，將結(jié)果實(shí)時(shí)輸出到數(shù)據(jù)庫(kù)中。