基于Spark的Canopy-FCM在氣象中的應(yīng)用

勾志竟，宮志宏，徐 梅，劉布春

(1.天津市氣象信息中心，天津 300074; 2.天津市氣候中心，天津 300074; 3.中國農(nóng)業(yè)科學(xué)院 農(nóng)業(yè)環(huán)境與可持續(xù)發(fā)展研究所，北京 100081)

0 引 言

隨著科技的進(jìn)步，氣象部門獲取數(shù)據(jù)的途徑也越來越多，收集并產(chǎn)生的氣象數(shù)據(jù)呈指數(shù)級增長。如何將數(shù)據(jù)挖掘技術(shù)應(yīng)用到氣象預(yù)報預(yù)測和氣象災(zāi)害預(yù)測等方面[1-3]，從海量的氣象數(shù)據(jù)中挖掘有價值的信息，成為氣象行業(yè)研究的重點(diǎn)。傳統(tǒng)的數(shù)據(jù)處理方法已經(jīng)不能很好地處理海量數(shù)據(jù)，挖掘數(shù)據(jù)內(nèi)部規(guī)律時更為乏力，而數(shù)據(jù)挖掘算法與分布式處理框架[4]的出現(xiàn)為挖掘海量氣象數(shù)據(jù)提供了一種新的思路。

陳正威[5]在Hadoop平臺上運(yùn)用預(yù)處理有向無環(huán)圖和支持向量機(jī)(PDAG-SVM)算法對降水量做出預(yù)測，該方法在預(yù)測精度和預(yù)測效率上都取得了令人滿意的結(jié)果；王昊等[6]提出了一種Hadoop平臺下基于離散貝葉斯網(wǎng)絡(luò)的數(shù)據(jù)挖掘改進(jìn)算法，預(yù)測精度明顯高于目前短期氣候預(yù)測中采用的樸素貝葉斯算法；張晨陽等[7]提出基于Hadoop的計算等價類的數(shù)據(jù)約簡算法與樸素貝葉斯分類算法，該并行數(shù)據(jù)挖掘方案可以有效處理海量氣象數(shù)據(jù)，并具有良好的擴(kuò)展性；Lv Zhenhua等[8]提出了并行K-means算法，并用于遙感圖像的分類；李莉等[9]基于Spark平臺提出并行K-means算法對氣候區(qū)進(jìn)行劃分，對氣象領(lǐng)域研究有重要現(xiàn)實(shí)意義。

從目前的相關(guān)研究可以看出，學(xué)者們不斷對海量數(shù)據(jù)挖掘方法進(jìn)行研究和優(yōu)化，而聚類算法作為數(shù)據(jù)挖掘的重要方法，將其與分布式處理框架相結(jié)合[10-12]處理海量數(shù)據(jù)成為數(shù)據(jù)挖掘領(lǐng)域越來越活躍的研究方向。文中提出了一種Canopy-FCM算法，可以有效避免模糊C-均值聚類算法對初始聚類中心敏感的問題，同時針對海量氣象數(shù)據(jù)，采用Spark內(nèi)存計算分布式框架快速有效地從氣象數(shù)據(jù)中挖掘出有用的信息，大大的提高了運(yùn)行效率。

1 模糊C均值算法(FCM)

模糊C均值(fuzzy C-means，F(xiàn)CM)算法[13]是1974年由Dunn提出并由Bezdek推廣的，它是基于模糊集合論，把聚類問題轉(zhuǎn)化為非線性規(guī)劃問題，并通過迭代求解。

令X={X1,X2,…,Xn}為待分類樣本，F(xiàn)CM將其分為c個模糊組，使得目標(biāo)函數(shù)值最小，目標(biāo)函數(shù)如下所示：

(1)

(2)

其中，uij是樣本j屬于類i的隸屬度，Ci為第i類的中心，m∈[1,∞]為模糊因子。

通過式(2)，采用拉格朗日乘數(shù)法構(gòu)造以下目標(biāo)函數(shù)：

(3)

對所有參數(shù)求導(dǎo)，得到使式(3)達(dá)到最小值的必要條件為:

(4)

(5)

由式(4)和式(5)可以知道，給定初始樣本集合X，以及分類數(shù)目c和模糊因子m，F(xiàn)CM算法按照以下步驟不斷迭代就可以計算出隸屬度矩陣U和聚類中心C。

(1)用隨機(jī)數(shù)生成器生成初始隸屬度矩陣U，且滿足約束條件式(2)。

(2)用式(4)更新聚類中心。

(3)用式(5)更新隸屬度矩陣U。

(4)計算式(1)的目標(biāo)函數(shù)值，如果小于閾值ε，則算法停止，否則重復(fù)步驟(2)和(3)。

2 Canopy-FCM算法設(shè)計

2.1 Canopy算法

FCM算法采用隨機(jī)生成聚類中心的方法，但無法保證為每個分類找到較好的中心，而聚類中心直接影響算法的運(yùn)行效率。針對初始中心敏感，容易陷入局部最優(yōu)的問題，文中采用Canopy算法[14]初始化聚類中心。Canopy算法可以很快得到最優(yōu)的分類數(shù)，其具體步驟如下：

(1)給定樣本X1,X2,…,Xn，設(shè)定初始閾值T1，T2，T1>T2。

(2)在樣本中隨機(jī)挑選樣本x，計算x到其他樣本點(diǎn)的距離d。

(3)把d

(4)重復(fù)步驟2、3，直到數(shù)據(jù)集為空。

2.2 Canopy-FCM算法框架

Canopy-FCM算法基本步驟如下:

Step1：利用Canopy算法生成初始聚類中心。

Step2：初始化隸屬度矩陣U。

Step3：更新聚類中心C。

Step4：更新隸屬度矩陣U。

Step5：是否滿足終止條件，若滿足，則算法停止；否則，重復(fù)Step3和Step4。

3 基于Spark的并行Canopy-FCM模型

3.1 Spark計算模型

Spark是基于內(nèi)存計算的分布式計算框架，起源于加利福尼亞大學(xué)伯克利分校的實(shí)驗(yàn)室研究項(xiàng)目[15]，其低延遲、低系統(tǒng)開銷、容錯性高、分布式數(shù)據(jù)結(jié)構(gòu)以及強(qiáng)大的函數(shù)式編程接口可以很好應(yīng)對迭代式計算應(yīng)用的高性能需求，在大規(guī)模數(shù)據(jù)處理任務(wù)中有廣泛的應(yīng)用。

Spark在分布式環(huán)境下采用主從結(jié)構(gòu)模型，包括Driver和Worker節(jié)點(diǎn)，程序運(yùn)行之前將數(shù)據(jù)存儲在Hadoop Distributed File System(HDFS)中，接著Driver會運(yùn)行應(yīng)用中的方法創(chuàng)建SparkContext以及RDD，DAGScheduler對象將每個job分成多個Stage，并為每個stage創(chuàng)建TaskSet，TaskScheduler將task提交給executor執(zhí)行，executor調(diào)用Taskrunner封裝task，并行線程池中取一個線程執(zhí)行task。其架構(gòu)如圖1所示。

圖1 Spark架構(gòu)

3.2 Canopy-FCM算法的并行化

基于Spark的Canopy-FCM算法流程如圖2所示。

圖2 基于Spark的Canopy-FCM算法流程

(1)配置好Spark運(yùn)行環(huán)境并初始化各參數(shù)。通過hadoop fs -put命令將數(shù)據(jù)上傳到HDFS上，調(diào)用SparkContext的sc.textFile()方法將數(shù)據(jù)轉(zhuǎn)換為Spark-RDD，通過map操作轉(zhuǎn)換為向量緩存到內(nèi)存中。

當(dāng)我們在進(jìn)行股票運(yùn)作時，追求的是績優(yōu)股，當(dāng)我們在購買房產(chǎn)時，追求的是未來的黃金地段，同樣，國有企業(yè)在進(jìn)行經(jīng)營時所追求的就是績優(yōu)股、黃金地段，就是資產(chǎn)的良性發(fā)展。油田企業(yè)作為國有企業(yè)的一員，就是要維護(hù)國有資產(chǎn)安全，保障企業(yè)可持續(xù)發(fā)展。從當(dāng)前企業(yè)審計的要求來看，質(zhì)量和責(zé)任是企業(yè)審計的基礎(chǔ)，績效是企業(yè)審計的方向和目標(biāo)，就是要發(fā)現(xiàn)企業(yè)是否成為了藍(lán)籌股，把投入產(chǎn)出比例如何作為國有資產(chǎn)的組成部分，油田企業(yè)的績效審計越來越被重視，它進(jìn)一步擴(kuò)展和深化了油田企業(yè)審計的內(nèi)涵。

(2)在各個子節(jié)點(diǎn)通過map操作計算數(shù)據(jù)集中每個點(diǎn)到Canopy中心點(diǎn)的歐氏距離，進(jìn)而得到局部的Canopy中心點(diǎn)，然后通過reduce操作得到全局的Canopy中心點(diǎn)，將其作為FCM的初始聚類中心，并廣播給各個子節(jié)點(diǎn)。

(3)在各個子節(jié)點(diǎn)通過map操作計算每個數(shù)據(jù)點(diǎn)到各中心的歐氏距離和隸屬度，然后通過reduceByKey()和collectAsMap()方法得到各數(shù)據(jù)點(diǎn)到每個分類的距離之和與隸屬度之和，對隸屬度和聚類中心進(jìn)行更新。

(4)計算目標(biāo)函數(shù)的值，判斷結(jié)果是否收斂，如果收斂則算法結(jié)束，通過Combine操作合并中間結(jié)果，并通過Reduce操作得到全局聚類中心，否則重復(fù)步驟(3)。

Canopy-FCM算法并行化[16]的偽代碼如下：

Input:X={X1,X2,…,Xn}，T1，T2，m，K

Output:C={C1,C2,…,Cc}

Initialization();

l←data.mapPartitions{points?

forxi←points{

}.reduce(merge)

C'←C;C←null

forj=1 toC{

(sumUX,sumU)←l(j)

C+=sumUX/sumU

}

}

4 實(shí)例分析

4.1 實(shí)驗(yàn)環(huán)境與數(shù)據(jù)集

實(shí)驗(yàn)采用Spark分布式集群，集群搭建在服務(wù)器虛擬化平臺上，選取1臺機(jī)器作為主節(jié)點(diǎn)，其他7臺機(jī)器作為工作節(jié)點(diǎn)。虛擬機(jī)各項(xiàng)配置及集群的配置信息分別如表1、表2所示，實(shí)驗(yàn)數(shù)據(jù)采用天津經(jīng)過質(zhì)控后的208個區(qū)域自動氣象站4～10月夏半年逐月降水觀測數(shù)據(jù)。

表1 虛擬機(jī)配置信息

由表2可以看出Spark分布式集群在運(yùn)行時需要一系列的后臺程序，主要有:

Master-負(fù)責(zé)資源的調(diào)度(決定在哪些Worker上執(zhí)行executor)和監(jiān)控Worker。

Worker-負(fù)責(zé)執(zhí)行任務(wù)的進(jìn)程(executor)，并將當(dāng)前機(jī)器的信息通過心跳匯報給Master。

NameNode-負(fù)責(zé)管理文件系統(tǒng)的Namespace。

DataNode-負(fù)責(zé)管理各個存儲節(jié)點(diǎn)。

SecondaryNameNode-NameNode的熱備，負(fù)責(zé)周期性地合并Namespace image和Edit log。

表2 集群信息配置

4.2 實(shí)驗(yàn)結(jié)果及分析

實(shí)驗(yàn)結(jié)果如圖3所示，由圖3可以看出天津208個區(qū)域自動氣象站降水分布可分為4個區(qū)域，1區(qū)主要集中在中部和北部區(qū)域，共有96個站；2區(qū)集中在東部區(qū)域，共有29個站；3區(qū)集中在東南部，共有31個站；4區(qū)主要集中在西南部，共有52個站。

圖3 天津降水區(qū)劃圖

圖4是實(shí)驗(yàn)得到的天津市4個分區(qū)降水量年平均分布圖，由圖4可以看出，4個分區(qū)的降水主要集中在6～9月，7月降水量最為顯著，其次是8月、6月、9月，這一趨勢與中國氣象局氣象數(shù)據(jù)中心發(fā)布的天津氣候類型圖(1981-2010)一致。4個分區(qū)的具體分析如下：

圖4 天津市4個分區(qū)降水量年平均分布

1區(qū)主要位于天津中部和北部區(qū)域，該區(qū)域土壤以鹽化潮土和粘質(zhì)土為主，5月年平均降水量遠(yuǎn)低于其他分區(qū)，4～10月總年平均降水量485.7 mm。

2區(qū)主要位于天津的東部區(qū)域，屬于海積、沖積平原區(qū)，地勢北高南低，4月、7月、10月年平均降水量高于其他三個分區(qū)，6月年平均降水量遠(yuǎn)低于其他分區(qū)，4～10月總年平均降水量498.1 mm。

3區(qū)主要位于天津市的東南沿海地區(qū)，地勢低平，以海積低平原為主，土層受海潮影響鹽漬化比較嚴(yán)重，5月、8月、9月年平均降水量遠(yuǎn)高于其他三個分區(qū)，4～10月總年平均降水量508.1 mm。

4區(qū)主要位于天津的西南部，該區(qū)域以洼地沖積平原和濱海平原為主，地形平坦但多洼地，地勢南高北低，西高東低，4月、7月、8月、9月及10月年平均降水量均低于其他分區(qū)，4～10月總年平均降水量425.2 mm。

為了對比文中設(shè)計的Spark平臺和Hadoop平臺的集群性能，分別在Hadoop環(huán)境下和Spark環(huán)境下由單節(jié)點(diǎn)到8節(jié)點(diǎn)執(zhí)行相同大小的區(qū)域自動站降水?dāng)?shù)據(jù)文件，得到兩種環(huán)境下的加速比，如圖5所示。

圖5 Hadoop平臺和Spark平臺的加速比

由圖5不難看出，當(dāng)節(jié)點(diǎn)數(shù)目為單節(jié)點(diǎn)時，Hadoop平臺和Spark平臺的性能都處于最差。隨著DataNode節(jié)點(diǎn)數(shù)量的增加，Spark平臺和Hadoop平臺算法的運(yùn)行時間都有不同程度縮短，而Spark平臺的加速比要優(yōu)于Hadoop平臺，說明文中提出的算法在Spark平臺下能有效地提高算法的性能，及時準(zhǔn)確地挖掘出海量氣象數(shù)據(jù)的有價值信息。

5 結(jié)束語

針對模糊C-均值聚類算法對初始聚類中心敏感及因迭代計算次數(shù)增加導(dǎo)致內(nèi)存不足的問題，設(shè)計了一種基于Spark框架的Canopy-FCM并行化聚類算法。該算法結(jié)合Canopy算法與模糊C-均值聚類算法，避免了FCM算法對初始化敏感的問題，并結(jié)合Spark分布式框架內(nèi)存計算的優(yōu)勢，大大降低了海量氣象數(shù)據(jù)的處理時間。通過采用天津市208個區(qū)域自動氣象站4～10月逐月降水觀測數(shù)據(jù)，評估了天津市不同區(qū)域的降水情況。實(shí)驗(yàn)結(jié)果表明，提出的方法不僅可以快速有效地從氣象數(shù)據(jù)中挖掘出有用的信息，同時還有良好的擴(kuò)展性，能夠?yàn)橄嚓P(guān)部門做好抗旱救災(zāi)、防災(zāi)救災(zāi)工作提供一種全新的思路和方法。但方法僅針對降水區(qū)進(jìn)行了劃分，未來可以結(jié)合溫度、濕度、干燥度等因素做進(jìn)一步的氣候區(qū)劃研究。