基于MapReduce 云計(jì)算的報(bào)表平臺(tái)技術(shù)與應(yīng)用

俞紅威 朱 建 蔡玉立

中博信息技術(shù)研究院有限公司

0 引言

云計(jì)算是分布式計(jì)算的一種，指的是通過(guò)網(wǎng)絡(luò)“云”將巨大的數(shù)據(jù)計(jì)算處理程序分解成無(wú)數(shù)個(gè)小程序，然后，通過(guò)多服務(wù)器組成的系統(tǒng)進(jìn)行處理和分析這些小程序得到結(jié)果并返回給用戶。云計(jì)算早期，就是簡(jiǎn)單的分布式計(jì)算，解決任務(wù)分發(fā)，并進(jìn)行計(jì)算結(jié)果的合并。因而，云計(jì)算又稱為網(wǎng)格計(jì)算。通過(guò)這項(xiàng)技術(shù)，可以在很短的時(shí)間內(nèi)完成對(duì)數(shù)以萬(wàn)計(jì)的數(shù)據(jù)的處理，從而能夠提供強(qiáng)大的網(wǎng)絡(luò)服務(wù)。

企業(yè)的業(yè)務(wù)數(shù)據(jù)在數(shù)據(jù)量越來(lái)越大的情況下，業(yè)務(wù)取數(shù)邏輯也越來(lái)越復(fù)雜。比如集團(tuán)型企業(yè)財(cái)務(wù)月報(bào)，單個(gè)財(cái)務(wù)報(bào)表戶涉及到8 張不同的報(bào)表，包括資產(chǎn)負(fù)債表、利潤(rùn)表、收入明細(xì)表等，共計(jì)近5000 個(gè)指標(biāo)，再加上集團(tuán)型企業(yè)涉及多財(cái)務(wù)報(bào)表戶，按照1000 個(gè)財(cái)務(wù)賬套計(jì)算則有500 萬(wàn)個(gè)指標(biāo)的計(jì)算，因此報(bào)表數(shù)據(jù)的計(jì)算量非常大，云計(jì)算的分布式計(jì)算特點(diǎn)特別適合這種場(chǎng)景的計(jì)算。

1 技術(shù)現(xiàn)狀

1.1 MapReduce 技術(shù)簡(jiǎn)介

云計(jì)算的關(guān)鍵技術(shù)是MapReduce，最早是由Google 提出，后來(lái)也運(yùn)用在開(kāi)源的云技術(shù)Hadoop 中。與傳統(tǒng)開(kāi)發(fā)模式相比，使用MapReduce 模式，開(kāi)發(fā)人員在拆解問(wèn)題的過(guò)程中，就要先對(duì)問(wèn)題進(jìn)行平行化處理。開(kāi)發(fā)人員需要先分析問(wèn)題的解決流程，找出可以利用平行運(yùn)算來(lái)處理數(shù)據(jù)的部分，也就是那些能夠被切成小段分開(kāi)來(lái)處理的數(shù)據(jù)，再針對(duì)可以采用平行處理的部分寫(xiě)成Map 程序。

MapReduce 本身源自于函數(shù)式語(yǔ)言，主要通過(guò)“Map（映射）”和“Reduce（化簡(jiǎn)）”這兩個(gè)步驟來(lái)并行處理大規(guī)模的數(shù)據(jù)集。首先，Map 會(huì)先對(duì)由很多獨(dú)立元素組成的邏輯列表中的每一個(gè)元素進(jìn)行指定的操作，且原始列表不會(huì)被更改，會(huì)創(chuàng)建多個(gè)新的列表來(lái)保存Map 的處理結(jié)果。當(dāng)Map 工作完成之后，系統(tǒng)會(huì)接著對(duì)新生成的多個(gè)列表進(jìn)行清理和排序，然后會(huì)對(duì)這些新創(chuàng)建的列表進(jìn)行Reduce 操作，也就是對(duì)一個(gè)列表中的元素根據(jù)Key 值進(jìn)行適當(dāng)?shù)暮喜?。圖1 是常見(jiàn)MapReduce 的體系結(jié)構(gòu)。

圖1 MapReduce 的體系結(jié)構(gòu)

1.2 ICE 網(wǎng)絡(luò)通信引擎技術(shù)簡(jiǎn)介

網(wǎng)絡(luò)通信引擎ICE（Internet Communications Engine）是Zero C公司的分布式系統(tǒng)開(kāi)發(fā)專家實(shí)現(xiàn)的一種新的，高性能的，面向?qū)ο笾虚g件平臺(tái)?；贗CE 可以實(shí)現(xiàn)電信級(jí)分布式網(wǎng)絡(luò)通信的解決方案。ICE 為構(gòu)建面向?qū)ο蟮目蛻?服務(wù)器應(yīng)用提供了工具、API 和庫(kù)支持。

ICE 作為中間件平臺(tái)，客戶端及服務(wù)端的應(yīng)用都是由應(yīng)用代碼及ICE 的庫(kù)代碼混合組成的?？蛻魬?yīng)用及服務(wù)器應(yīng)用分別對(duì)應(yīng)用的是客戶端與服務(wù)端。代理是根據(jù)SLICE 定義的ICE 文件實(shí)現(xiàn)，它提供了一個(gè)向下調(diào)用的接口，提供了數(shù)據(jù)的序列化與反序化。

2 總體技術(shù)架構(gòu)

2.1 方案介紹

考慮到場(chǎng)景的差異和實(shí)際項(xiàng)目需要，本方案沒(méi)有直接采用Hadoop 的MapReduce架構(gòu)。而是采用云計(jì)算MapReduce 的思想，自主開(kāi)發(fā)云計(jì)算分布式框架，將報(bào)表主任務(wù)進(jìn)行分解，調(diào)用任務(wù)的Mapper分解程序，根據(jù)不同的取數(shù)公式及參數(shù)拆分成若干個(gè)子任務(wù)。系統(tǒng)初始化一個(gè)子任務(wù)池，定義任意個(gè)取數(shù)線程，輪詢獲取分解的子任務(wù)，通過(guò)ICE RPC 遠(yuǎn)程調(diào)用方法實(shí)現(xiàn)任務(wù)遠(yuǎn)程調(diào)用，將子任務(wù)分配到遠(yuǎn)程計(jì)算節(jié)點(diǎn)，計(jì)算完成后返回此子任務(wù)的結(jié)果，并且將結(jié)果進(jìn)行匯總，當(dāng)所有的子任務(wù)都已經(jīng)全部執(zhí)行成功后，啟動(dòng)Reduce 程序，針對(duì)匯總的結(jié)果進(jìn)行指標(biāo)計(jì)算，并保存到數(shù)據(jù)庫(kù)中，從而完成報(bào)表主任務(wù)的計(jì)算。圖2 為整體方案架構(gòu)圖。

圖2 整體方案架構(gòu)圖

配套整個(gè)系統(tǒng)設(shè)計(jì)，系統(tǒng)還包括表套，表單，指標(biāo)，取數(shù)公式，報(bào)表戶，時(shí)間維度（年度，月度）等基本信息。表套是由一系列相關(guān)的表單組成的，表單則由若干指標(biāo)組成，它包括名稱，編碼和具體的取數(shù)公式構(gòu)成。取數(shù)公式可能包括若干公式，通過(guò)加減乘除等四則運(yùn)算計(jì)算而來(lái)。系統(tǒng)的計(jì)算任務(wù)是指具體一個(gè)或多個(gè)報(bào)表戶的某年某月某個(gè)表套或者其中多個(gè)表單的取數(shù)任務(wù)。

2.2 方案優(yōu)勢(shì)

系統(tǒng)采用多線程高并發(fā)的計(jì)算模式，通過(guò)負(fù)載均衡策略實(shí)現(xiàn)了網(wǎng)格計(jì)算，極大提高了系統(tǒng)的計(jì)算性能。具體的優(yōu)勢(shì)包括下面幾個(gè)方面：

基于MapReduce 高性能并發(fā)計(jì)算框架；

支持取數(shù)線程池的任意動(dòng)態(tài)擴(kuò)展；

支持遠(yuǎn)程計(jì)算節(jié)點(diǎn)的任意動(dòng)態(tài)擴(kuò)展；

可靠的輪詢調(diào)度機(jī)制，確保遠(yuǎn)程節(jié)點(diǎn)任務(wù)均衡分配；

高效可靠的RPC 遠(yuǎn)程調(diào)用機(jī)制；

可靠的容錯(cuò)機(jī)制，在由于網(wǎng)絡(luò)異常情況保證任務(wù)的成功調(diào)用。

3 詳細(xì)解決方案

3.1 報(bào)表定義（Report Define）

基礎(chǔ)設(shè)置部分由用戶首先通過(guò)表樣定義，指標(biāo)生成，公式定義等操作，將指標(biāo)編碼、指標(biāo)名稱和指標(biāo)取數(shù)表達(dá)式進(jìn)行初始化編制，具體的取數(shù)公式需要對(duì)應(yīng)開(kāi)發(fā)人員開(kāi)發(fā)相對(duì)應(yīng)的取數(shù)插。為能夠說(shuō)明本方案關(guān)于報(bào)表任務(wù)基于MapReduce的過(guò)程，本文章中僅僅截取報(bào)表中的部分指標(biāo)作為舉例描述。生成出的部分表樣及取數(shù)表達(dá)式如表1 所示。

表1 指標(biāo)編碼名稱公式

表1 中共包括6 個(gè)指標(biāo)，其中有4 個(gè)指標(biāo)是正常的取數(shù)指標(biāo)（[443_E_1]， [443_E_2]， [443_F_1]， [443_F_1]），兩個(gè)指標(biāo)是表間運(yùn)算的指標(biāo)（[443_E_3]，[443_F_3]）。將指標(biāo)的公式信息一直迭代到最后取數(shù)公式為止，相當(dāng)于將[443_E_3]指標(biāo)的取數(shù)表達(dá)式變更成UDEF_FYD（“y”，“6001”）-UDEF_FYD（“y”，“600190”）+UDEF_FYD（“y”，“6002”）-UDEF_FYD（“y”，“600192”）。其中UDEF_FYD 和UEDF_ND 注冊(cè)成指定的JAVA 類，平臺(tái)要求將注冊(cè)的取數(shù)Java 文件編譯后打包成jar 文件。

3.2 任務(wù)分解（Task Mapper）

任務(wù)分解是系統(tǒng)中的核心步驟，用來(lái)將復(fù)雜任務(wù)分解成多個(gè)簡(jiǎn)單任務(wù)，這樣會(huì)使得任務(wù)能夠并行計(jì)算。系統(tǒng)中TaskMapper 程序負(fù)責(zé)報(bào)表任務(wù)的MAP 分解過(guò)程。提供一個(gè)主線程，專門(mén)掃描待處理表中的未處理的報(bào)表任務(wù)，得到報(bào)表任務(wù)后，會(huì)根據(jù)包含的表套或表單得到具體需要計(jì)算的指標(biāo)，并循環(huán)指標(biāo)列表，直接從緩存中拿到指標(biāo)包含的取數(shù)公式及參數(shù)信息。根據(jù)上面示例中得到包含的取數(shù)公式數(shù)據(jù)是UDEF_FYD（發(fā)生額），參數(shù)為（"y"，"6001"），（"y"，"600190"），（"y"，"6002"），（"y"，"600192"）。UDEF_ND（累計(jì)額）公式，參數(shù)包括（"y"，"6001"），（"y"，"600190"），（"y"，"6002"），（"y"，"600192"）。此任務(wù)中包括兩個(gè)具體公式的業(yè)務(wù)數(shù)據(jù)取數(shù)，UDEF_FYD（發(fā)生額）和UDEF_ND（累計(jì)額），系統(tǒng)會(huì)為這兩個(gè)公式分別定義報(bào)表戶分解數(shù)和參數(shù)分解數(shù)，并根據(jù)這兩個(gè)參數(shù)來(lái)分解子任務(wù)。如果參數(shù)分別設(shè)成0，那么就是有幾個(gè)具體公式就生成幾個(gè)任務(wù)，否則會(huì)根據(jù)這兩個(gè)參數(shù)嵌套循環(huán)分解子任務(wù)。任務(wù)分解后，會(huì)將子任務(wù)加到報(bào)表子任務(wù)池中。如表2 所示，是不同參數(shù)配置下某個(gè)取數(shù)公式任務(wù)分解說(shuō)明以及生成的子任務(wù)個(gè)數(shù)舉例。

表2 不同參數(shù)配置下取數(shù)公式任務(wù)分解說(shuō)明

假設(shè)此時(shí)只計(jì)算1 個(gè)報(bào)表戶，報(bào)表戶編碼為A3201，同時(shí)UDEF_FYD 和UDEF_ND 兩個(gè)公式的賬戶分解數(shù)為0，參數(shù)分解數(shù)分別為4 和2 時(shí)，根據(jù)上述報(bào)表定義的指標(biāo)和取數(shù)公式配置共分解成以下3 個(gè)子任務(wù)，其中UDEF_FYD 公式有1 個(gè)任務(wù)，UDEF_ND 公式有2 個(gè)任務(wù)，具體任務(wù)、公式及參數(shù)如表3 所示。

表3 任務(wù)拆分的具體公式和參數(shù)

通過(guò)對(duì)報(bào)表指標(biāo)公式表達(dá)式的解析和分解，將指標(biāo)計(jì)算任務(wù)按照公式的維度進(jìn)行分組，并且根據(jù)參數(shù)進(jìn)行分解，這樣將一個(gè)大的報(bào)表任務(wù)分解成任意多個(gè)以公式為維度的子任務(wù)。從以指標(biāo)維度進(jìn)行同步橫向計(jì)算變成了按照公式維度進(jìn)行縱向并行計(jì)算。這些子任務(wù)通過(guò)并行計(jì)算，改變了整個(gè)報(bào)表計(jì)算的模式。

3.3 任務(wù)輪詢分配（Task Polling）

當(dāng)將報(bào)表任務(wù)分解成若干個(gè)子任務(wù)后，將子任務(wù)加入到子任務(wù)池。系統(tǒng)定義了多個(gè)取數(shù)線程數(shù)，系統(tǒng)初始化時(shí)，會(huì)啟動(dòng)定義的取數(shù)線程，輪詢從子任務(wù)池中獲取一個(gè)任務(wù)，將任務(wù)更新成處理中狀態(tài)。

遠(yuǎn)程計(jì)算節(jié)點(diǎn)定義心跳服務(wù)端口，并且提供服務(wù)端狀態(tài)更新程序調(diào)用，定期將不可用的節(jié)點(diǎn)移除出可分配遠(yuǎn)程計(jì)算節(jié)點(diǎn)，確保任務(wù)分配成功率。遠(yuǎn)程計(jì)算節(jié)點(diǎn)啟動(dòng)時(shí)會(huì)根據(jù)配置的端口啟動(dòng)心跳服務(wù)，提供服務(wù)端調(diào)用，服務(wù)端調(diào)用成功后將此節(jié)點(diǎn)更新成可用狀態(tài)，如果無(wú)法訪問(wèn)則標(biāo)記成不可訪問(wèn)，具體方案如圖3 任務(wù)輪詢機(jī)制示意圖所示。

圖3 任務(wù)輪詢機(jī)制示意圖

本方案中輪詢調(diào)用算法采用權(quán)重輪詢調(diào)度算法 （Weighted Round-Robin Scheduling）。權(quán)重輪詢調(diào)度算法可以解決服務(wù)器間性能不一的情況，它用相應(yīng)的權(quán)值表示服務(wù)器的處理性能，按權(quán)值的高低和輪詢方式分配請(qǐng)求到各服務(wù)器。權(quán)重輪詢算法根據(jù)新建連接數(shù)來(lái)調(diào)度，權(quán)值高的服務(wù)器先收到連接，權(quán)重值越高被輪詢到的次數(shù)（概率）也越高，相同權(quán)值的服務(wù)器處理相同數(shù)目的連接數(shù)。

3.4 任務(wù)遠(yuǎn)程調(diào)用（RPC Invoke）

當(dāng)通過(guò)任務(wù)輪詢分配，獲得可用的遠(yuǎn)程計(jì)算節(jié)點(diǎn)之后，就需要通過(guò)遠(yuǎn)程調(diào)用方式，將需要計(jì)算的子任務(wù)發(fā)送到遠(yuǎn)程計(jì)算節(jié)點(diǎn)。首先需要定義子任務(wù)信息，子任務(wù)包括公式代碼、jar 包文件名、年份、月份、定義JAVA 類、拆分的調(diào)用參數(shù)、報(bào)表戶等信息。然后通過(guò)RPC 遠(yuǎn)程調(diào)用發(fā)送子任務(wù)，在本方案中，采用ICE 作為RPC 框架，負(fù)責(zé)底層傳輸調(diào)用，技術(shù)人員關(guān)注業(yè)務(wù)邏輯的編寫(xiě)。

RPC（Remote Producer Call），指遠(yuǎn)程過(guò)程調(diào)用，Remote Producer Call Protocol 是一種計(jì)算機(jī)通信協(xié)議。它允許像調(diào)用本地方法一樣調(diào)用遠(yuǎn)程服務(wù)。首先根據(jù)ICE 技術(shù)架構(gòu)要求，其傳輸核心代碼是通過(guò)對(duì)SLICE 文件解析動(dòng)態(tài)生成。系統(tǒng)定義的SLICE 文件如下，文件定義了1 個(gè)調(diào)用方法call，在消息體中定義同步和異步標(biāo)識(shí)。

3.5 任務(wù)匯聚（Task Reducer）

通過(guò)遠(yuǎn)程同步調(diào)用后，在計(jì)算節(jié)點(diǎn)上動(dòng)態(tài)生成任務(wù)信息中的指定JAVA 類的方法。方案中要求指定的JAVA 類繼承系統(tǒng)提供RPFormulaBase 基礎(chǔ)類，動(dòng)態(tài)加載對(duì)象實(shí)例，根據(jù)公式包含的參數(shù)，以及調(diào)用參數(shù)，構(gòu)造String[]參數(shù)，從而反射調(diào)用calculate（String[] p1，String[] p2，……，String[] pn）方法。此方法會(huì)返回Object 對(duì)象。在calculate 方法中，可以根據(jù)報(bào)表戶、年月、調(diào)用參數(shù)這些參數(shù)，從相應(yīng)的業(yè)務(wù)表中執(zhí)行邏輯，生成需要的取數(shù)結(jié)果。MAP 對(duì)象中KEY 的形式為“報(bào)表戶編碼_公式編碼（參數(shù)1、參數(shù)2……參數(shù)n）”格式。

當(dāng)所有的子任務(wù)全部成功執(zhí)行并返回結(jié)果后，會(huì)在服務(wù)端匯聚成一個(gè)包含全部返回結(jié)果的MAP 對(duì)象，然后會(huì)啟動(dòng)指標(biāo)結(jié)果的Reduce 過(guò)程。如表4 所示，描述了3 個(gè)任務(wù)的基本信息以及分別反饋回來(lái)的MAP 匯總后的KEY 和VALUE。

表4 任務(wù)反饋后MAP 中對(duì)應(yīng)KEY 和VALUE 值

在得到全量MAP 的信息之后，通過(guò)循環(huán)所有的指標(biāo)，在系統(tǒng)緩存中快速得到指轉(zhuǎn)換后取數(shù)公式，通過(guò)KEY 值檢索到對(duì)應(yīng)的VALUE，替換取數(shù)公式中的原公式和參數(shù)信息。如指標(biāo)[443_E_1]的取數(shù)表達(dá)式為UDEF_FYD（"y"，"6001"）-UDEF_FYD（"y"，"600190"），從指標(biāo)緩存中得到包含的取數(shù)公式、參數(shù)及報(bào)表戶編碼，形成KEY 的值為A3201_UDEF_FYD（"y"，"6001"）、A3201_UDEF_FYD（"y"，"600190"）。假設(shè)從全局MAP 緩存中根據(jù)KEY 值分別得到400、200，分別用這二個(gè)值進(jìn)行替換得到最終的取數(shù)表達(dá)式為“（400）-（200）”，再通過(guò)引擎計(jì)算得到值為200，全部指標(biāo)進(jìn)行替換后形成如表5 所示的數(shù)值。

表5 全部指標(biāo)經(jīng)過(guò)引擎計(jì)算后指標(biāo)數(shù)值

當(dāng)?shù)玫饺恐笜?biāo)的值后，并且通過(guò)SQL 批量插入方式進(jìn)行批量更新，從而完成指標(biāo)計(jì)算結(jié)果的保存，最終完成任務(wù)匯聚過(guò)程。系統(tǒng)會(huì)更新主任務(wù)狀態(tài)為計(jì)算完成狀態(tài)，通過(guò)前端頁(yè)面任務(wù)主動(dòng)提醒用戶已經(jīng)完成報(bào)表任務(wù)的計(jì)算。

4 結(jié)束語(yǔ)

本文針對(duì)集團(tuán)級(jí)復(fù)雜報(bào)表的計(jì)算提供了一種新的基于MapReduce 云計(jì)算計(jì)算報(bào)表平臺(tái)解決方案，能夠解決大數(shù)據(jù)規(guī)模下報(bào)表取數(shù)的性能問(wèn)題，通過(guò)對(duì)報(bào)表任務(wù)的分解，將報(bào)表任務(wù)按照取數(shù)公式分解成若干個(gè)子任務(wù)，運(yùn)用分布式計(jì)算處理子任務(wù)，完成子任務(wù)遠(yuǎn)程計(jì)算節(jié)點(diǎn)的遠(yuǎn)程調(diào)用。然后對(duì)全部結(jié)果數(shù)據(jù)進(jìn)行Reduce 操作。通過(guò)這種云計(jì)算分布式報(bào)表平臺(tái)的設(shè)計(jì)和使用，大大提升了大數(shù)據(jù)下報(bào)表計(jì)算的效率。當(dāng)然，整個(gè)方案還有很大的優(yōu)化點(diǎn)和提升空間，比如現(xiàn)有任務(wù)輪詢調(diào)度機(jī)制使用的權(quán)重輪詢調(diào)度算法，但此算法尚未考慮不同任務(wù)自身計(jì)算的復(fù)雜度，這些都需要在后續(xù)的研發(fā)過(guò)程中進(jìn)一步優(yōu)化和更新。下一步會(huì)將此方案運(yùn)用到其他符合MapReduce分布式計(jì)算的場(chǎng)景中。