基于Flink 框架的K-means 算法優(yōu)化及并行計(jì)算策略?

李召鑫 孟祥印 肖世德 胡鍇灃 賴煥杰

（西南交通大學(xué)機(jī)械工程學(xué)院 成都 610031）

1 引言

進(jìn)入到新世紀(jì)以來，伴隨著科學(xué)技術(shù)的飛速發(fā)展，在各個(gè)領(lǐng)域每時(shí)每刻都在產(chǎn)生成千上萬的數(shù)據(jù)，如何能高效地從這些海量數(shù)據(jù)中提取出有應(yīng)用價(jià)值的信息已成為當(dāng)前的研究熱點(diǎn)。K-means 是數(shù)據(jù)挖掘技術(shù)中解決海量數(shù)據(jù)聚類問題的經(jīng)典算法［1］，因其存在執(zhí)行快速、易于并行化等優(yōu)點(diǎn)故在許多領(lǐng)域都得到了廣泛的應(yīng)用。但同時(shí)存在一些不足：需指定聚類中心數(shù)；初始聚類中心的選取策略是完全隨機(jī)，這可能會(huì)影響到最終聚類結(jié)果的準(zhǔn)確率及計(jì)算速度。這兩種缺點(diǎn)嚴(yán)重限制了K-means算法聚類效果和計(jì)算效率的提高。

針對(duì)傳統(tǒng)K-means算法存在的不足，許多研究人員都提出了各種改進(jìn)算法，主要是對(duì)于聚類中心數(shù)的確定及聚類中心的選擇策略的優(yōu)化。杜佳穎等［2］提出了一種基于K-means++算法優(yōu)化初始中心點(diǎn)選擇的改進(jìn)算法，通過計(jì)算平均輪廓系數(shù)來指定K 值，該算法有效提高了計(jì)算效率；李淋淋等［3］提出了一種基于Spark 平臺(tái)的SBTICK-means 算法，針對(duì)傳統(tǒng)K-means算法在大規(guī)模數(shù)據(jù)計(jì)算時(shí)存在的瓶頸，引入Canopy 算法和三角形不等式定理，顯著提高了聚類結(jié)果速度及準(zhǔn)確率，并具有良好的擴(kuò)展性；梁彥［4］提出了一種基于Spark 平臺(tái)的并行K-means算法和Canopy K-means算法，該算法并行化后具有較好的聚類效果，相比于Hadoop 平臺(tái)，具有更高的效率；許明杰等［5］提出一種Spark 并行化的PSOK-means 算法，將PSO 算法與K-means 算法相結(jié)合，利用改進(jìn)后的PSO 算法提高K-means的全局搜索能力，并在Spark框架上進(jìn)行了實(shí)現(xiàn)，通過疾病檢測(cè)數(shù)據(jù)進(jìn)行聚類實(shí)驗(yàn)來驗(yàn)證所提出算法的效率；王義武等［6］提出了一種OCC K-means 算法，采用最大最小距離算法和UPGMA 來篩選初始中心，并基于Spark 框架進(jìn)行實(shí)現(xiàn)，有效提高了聚類精度及和計(jì)算速度。王美琪等［7］提出一種APMMD 算法，通過近鄰傳播算法獲取候選中心點(diǎn)，然后利用最大最小距離算法再選取k 個(gè)初始中心點(diǎn)，該算法有效減少了迭代次數(shù)。

Apache Flink［8～11］是近年來新興的一個(gè)大數(shù)據(jù)處理框架，與Hadoop不同，F(xiàn)link是一個(gè)集批處理和流處理于一體的計(jì)算框架，具有同時(shí)支持高吞吐、低延遲、高性能的優(yōu)點(diǎn)，現(xiàn)在的絕大部分聚類算法并行化研究大都基于批處理框架，如Hadoop、Spark［12～13］等，而基于Flink 實(shí)現(xiàn)的聚類算法研究較少?；谝陨项A(yù)研究，本文綜合上述聚類優(yōu)化算法的優(yōu)點(diǎn)，充分利用Flink框架在計(jì)算方面的優(yōu)勢(shì)，提出一種基于Flink 框架對(duì)K-means 優(yōu)化算法進(jìn)行并行化加速的方案，有效提升了K-means算法的聚類速度。

2 相關(guān)理論和技術(shù)

2.1 K-means算法

K-means 算法［14］（又稱為K-均值算法）是Mac-Queen在1967年提出的一種基于劃分的聚類算法，因其算法思想簡(jiǎn)單且易于實(shí)現(xiàn)被廣泛應(yīng)用于數(shù)據(jù)挖掘領(lǐng)域。該算法的作用將數(shù)據(jù)集S中的n個(gè)點(diǎn)劃分到K 個(gè)類中，具體操作步驟如下：先從數(shù)據(jù)集S中隨機(jī)抽取K 個(gè)點(diǎn)作為初始聚類中心C（C1，C2，…，Ck），然后計(jì)算數(shù)據(jù)集中每個(gè)樣本點(diǎn)X 到每個(gè)聚類中心的距離（一般采用歐式距離［15］式（1）作為度量）。

其中：d 表示兩個(gè)樣本點(diǎn)間的距離，n 為數(shù)據(jù)集的維度數(shù)量。

以此作為分類依據(jù)，按照最近距離公式將數(shù)據(jù)集中的各個(gè)樣本點(diǎn)X 分配給距離最近的聚類中心C（C1，C2，…，Ck），隨即更新每個(gè)聚類中心的值，設(shè)為其類別所有樣本的平均值，這個(gè)過程會(huì)不斷迭代直到達(dá)到迭代次數(shù)或聚類算法的誤差達(dá)到設(shè)定的閾值。

在聚類算法中，我們通常使用SSE（The sum of squares due to error）即誤差平方和來表示聚類的誤差大小，根據(jù)此值的大小來決定迭代是否結(jié)束。其公式如下：

其中：K 為聚類類別數(shù)，X 為數(shù)據(jù)集中的一個(gè)數(shù)據(jù)點(diǎn)，p為一個(gè)聚類中心。

2.2 Aphche Flink

Flink是Apache基金會(huì)旗下的一個(gè)開源的分布式計(jì)算框架，可同時(shí)支持流處理和批數(shù)據(jù)處理。Flink 和Spark 在架構(gòu)上有一定的相似性，都是基于Master-Slave 架構(gòu)，并且支持內(nèi)存計(jì)算。Flink 的數(shù)據(jù)流執(zhí)行圖包含以下三個(gè)階段（如圖1 所示）：Source、Transformation 和Sink。其中，Source 階段的作用是用來加載數(shù)據(jù)集，Transformation階段的作用是利用各種算子對(duì)數(shù)據(jù)流進(jìn)行相應(yīng)處理，主要包括map、flatMap、keyBy、reduce 等算子，Source 和Transformation 的并行度可以通過setParallelism 函數(shù)進(jìn)行自定義設(shè)置。Sink 階段負(fù)責(zé)數(shù)據(jù)處理結(jié)果的輸出，其并行度為1。

圖1 Flink框架工作架構(gòu)圖

3 基于Flink 的K-means 算法優(yōu)化策略

3.1 聚類類別K值的確定

K-means 算法需要指定聚類類別數(shù)K，K 的取值將直接影響到最終的聚類結(jié)果及準(zhǔn)確率，所以預(yù)先得到合適的K 值顯得尤為重要。Canopy 算法［16］是McCallum 在2000 年提出的一種聚類算法，該算法目的是將整個(gè)數(shù)據(jù)集粗略地劃分為不同的類別，優(yōu)點(diǎn)是不用預(yù)先設(shè)定K值就可以完成粗聚類，因此可用于K-means聚類之前，將粗聚類結(jié)果的結(jié)果作為K-means 的參照，避免了K 值選擇的盲目性，從而提高聚類速度。其流程如圖2所示。

圖2 Canopy算法確定K值流程圖

3.2 初始聚類中心的確定

初始聚類中心的理想狀態(tài)是分布盡量分散，這樣可以降低聚類結(jié)果陷入局部最優(yōu)的概率，而且可以提高計(jì)算效率。本文關(guān)于聚類中心的指定是基于最大距離算法進(jìn)行實(shí)現(xiàn)的，其基本思想如下，先輸入數(shù)據(jù)集和類別數(shù)量K，然后隨機(jī)選取一個(gè)數(shù)據(jù)點(diǎn)作為聚類中心，在選取剩下的K-1 個(gè)聚類中心時(shí)，通過比較數(shù)據(jù)集中的數(shù)據(jù)與已被選取的聚類中心的距離來找到最大距離dmax所對(duì)應(yīng)的數(shù)據(jù)點(diǎn)，然后將該點(diǎn)加入聚類中心，最終得到所有的聚類中心。這種方法可以解決K-means 算法隨機(jī)選取聚類中心時(shí)聚類中心過于鄰近的問題，從而減少迭代次數(shù)，有效提高算法效率。算法流程如圖3所示。

土地深松作業(yè)效果顯著。從2011年開始，河南省農(nóng)機(jī)深松整地工作由點(diǎn)到面，穩(wěn)步推進(jìn)。目前，全省累計(jì)開展農(nóng)機(jī)深松整地8300多萬畝，其中安排作業(yè)補(bǔ)助資金近7億元，累計(jì)補(bǔ)助作業(yè)面積2700余萬畝。各地農(nóng)機(jī)部門在技術(shù)模式、保障手段、監(jiān)督管理等方面開展了有益探索。通過示范帶動(dòng)，調(diào)動(dòng)了廣大農(nóng)民的土地深松積極性；通過技術(shù)培訓(xùn)，培養(yǎng)了一大批掌握農(nóng)機(jī)深松整地作業(yè)技術(shù)、帶頭實(shí)施深松作業(yè)的農(nóng)機(jī)人員；經(jīng)過試驗(yàn)研究，建立了適應(yīng)河南土壤特點(diǎn)、種植結(jié)構(gòu)的農(nóng)機(jī)深松整地作業(yè)技術(shù)模式和機(jī)具配置模式；根據(jù)基層實(shí)際，探索總結(jié)了組織作業(yè)、面積確認(rèn)、質(zhì)量管控和資金兌付的行之有效的工作措施。

圖3 最大距離算法確定聚類中心流程圖

3.3 基于Flink的K-means優(yōu)化算法并行化實(shí)現(xiàn)

Flink 計(jì)算框架本身支持并行計(jì)算，其并行度指的是同時(shí)運(yùn)行的線程的個(gè)數(shù)，通常一個(gè)Flink 程序由多個(gè)任務(wù)組成，這根據(jù)用戶設(shè)定的分組及并行度來確定。Flink 集群中的每個(gè)TaskManager 都包括多個(gè)slot，每個(gè)TaskManager 所擁有的slot 數(shù)量代表了該TaskManager 具有的并發(fā)執(zhí)行能力，此參數(shù)可以通過taskmanager.numberOfTaskSlots 參數(shù)來進(jìn)行設(shè)置，通常與該節(jié)點(diǎn)CPU 可用核數(shù)成正比。Flink 在某些條件下可以減少節(jié)點(diǎn)之間通信的開銷，是基于一種叫做任務(wù)鏈的技術(shù)來進(jìn)行實(shí)現(xiàn)的，當(dāng)多個(gè)算子設(shè)置了相同的并行度，而且是one to one 操作，這樣相連的算子可以鏈接在一起形成一個(gè)task，算子之間通過本地轉(zhuǎn)發(fā)（local forward）進(jìn)行連接，其邏輯視圖、并行化視圖、優(yōu)化后視圖如圖4所示。

圖4 Flink任務(wù)鏈?zhǔn)疽鈭D

本文算法所采取并行化策略如下，首先，在JobManager節(jié)點(diǎn)獲取到數(shù)據(jù)集之后，將其廣播到每個(gè)TaskManager 節(jié)點(diǎn)，采用最大距離算法逐次選取K 個(gè)初始聚類中心，再通過withBroadcastSet 函數(shù)將初始聚類中心的信息廣播到各個(gè)TaskManager 節(jié)點(diǎn)，在節(jié)點(diǎn)上迭代執(zhí)行K-means 算法，最后根據(jù)設(shè)置的迭代次數(shù)或者損失函數(shù)SSE 來判斷是否結(jié)束聚類過程。Flink 程序分為Source、Transformation、Sink三個(gè)階段，其中Source和Transformation階段是可以實(shí)現(xiàn)并行化的，通過并行化來減少數(shù)據(jù)加載及數(shù)據(jù)計(jì)算的過程，從而提高Flink 程序整體的運(yùn)行速度。JobManager主節(jié)點(diǎn)的作用是獲取數(shù)據(jù)集，并通過廣播分發(fā)至每個(gè)TaskManager，然后TaskManager迭代計(jì)算聚類中心到數(shù)據(jù)集中每個(gè)點(diǎn)的距離，將每一個(gè)數(shù)據(jù)點(diǎn)都?xì)w類到一個(gè)類別，最后將計(jì)算結(jié)果返回給JobManager，完成聚類結(jié)果的輸出。

4 實(shí)驗(yàn)環(huán)境與結(jié)果分析

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

本文實(shí)驗(yàn)平臺(tái)采用開源Flink 分布式框架，分別安裝在5 臺(tái)虛擬機(jī)上，由一臺(tái)JobManager 和4 臺(tái)TaskManager 組成。每一臺(tái)機(jī)器的配置都是：內(nèi)存2GB，硬盤20G，操作系統(tǒng)均為Centos 7.4。每一臺(tái)機(jī)器的軟件環(huán)境如表1所示。

表1 集群主要軟件環(huán)境詳情

本文實(shí)驗(yàn)采用的數(shù)據(jù)集是從加州大學(xué)UCI 實(shí)驗(yàn)室提供的機(jī)器學(xué)習(xí)數(shù)據(jù)庫中下載的iris、glass 和wine數(shù)據(jù)集，數(shù)據(jù)集詳情如表2所示。

表2 數(shù)據(jù)集相關(guān)信息

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

4.2.1 準(zhǔn)確率實(shí)驗(yàn)

為了對(duì)比傳統(tǒng)K-means 算法和本文算法的準(zhǔn)確率，本文通過對(duì)3 個(gè)不同屬性的數(shù)據(jù)集進(jìn)行聚類測(cè)試（對(duì)每種算法進(jìn)行20 次隨機(jī)實(shí)驗(yàn)取平均值），并統(tǒng)計(jì)聚類效果，聚類結(jié)果如表3 和表4 所示。從表3 和表4 可以看出，本文改進(jìn)算法在求解iris、glass 和wine 數(shù)據(jù)集時(shí)相比傳統(tǒng)K-means 算法的準(zhǔn)確率都有一定的提升，這表明本文基于Canopy 算法和最大距離算法的K-means 優(yōu)化算法減少了聚類過程的迭代次數(shù)，對(duì)計(jì)算效率有一定的提升。

表3 K-means算法與本文改進(jìn)算法準(zhǔn)確率對(duì)比

表4 K-means算法與本文改進(jìn)算法迭代次數(shù)對(duì)比

加速比是衡量并行化程序性能好壞的重要指標(biāo)，該指標(biāo)指的是同一個(gè)計(jì)算任務(wù)在單機(jī)環(huán)境和并行環(huán)境運(yùn)行消耗時(shí)間的比值，一般來說，加速比的值越大，代表算法并行加速的性能越好。為了測(cè)試本文算法在Flink 框架下的并行化能力，本文采用了synthetic_control數(shù)據(jù)集來進(jìn)行實(shí)驗(yàn)驗(yàn)證，該數(shù)據(jù)集含有600 個(gè)樣本點(diǎn)，因?yàn)樵瓟?shù)據(jù)集數(shù)據(jù)量較少，所以本文對(duì)原數(shù)據(jù)集進(jìn)行了擴(kuò)展，分別生成了含有3×10^5、3×10^6 和3×10^7 個(gè)樣本點(diǎn)的數(shù)據(jù)集，測(cè)試了不同節(jié)點(diǎn)數(shù)的加速比，實(shí)驗(yàn)結(jié)果如圖5所示。

圖5 基于Flink框架的改進(jìn)K-means算法的加速比

由圖5 可以看出，當(dāng)數(shù)據(jù)量比較小時(shí)，隨著節(jié)點(diǎn)數(shù)的增加，加速比折線趨于平緩，因?yàn)楫?dāng)數(shù)據(jù)集的規(guī)模較小時(shí)，計(jì)算時(shí)間比較短，如果節(jié)點(diǎn)數(shù)增加到一定數(shù)量后，會(huì)使得節(jié)點(diǎn)間的數(shù)據(jù)傳輸和任務(wù)調(diào)度等其他方面的時(shí)間開銷變大，進(jìn)而降低算法效率。相反，當(dāng)數(shù)據(jù)集達(dá)到一定規(guī)模后，加速比與節(jié)點(diǎn)數(shù)近似呈線性關(guān)系，實(shí)驗(yàn)可以表明本文算法在處理大規(guī)模的數(shù)據(jù)時(shí)具有較高的計(jì)算效率。

5 結(jié)語

本文針對(duì)傳統(tǒng)K-means 算法在處理海量數(shù)據(jù)的過程中在選取初始聚類中心時(shí)存在的易陷入局部最優(yōu)解和計(jì)算速度較慢等問題，提出一種基于Flink 框架并行化的K-means 優(yōu)化算法。該算法采取Canopy 算法來計(jì)算K-means 算法輸入所必需的K 值，并利用最大距離算法優(yōu)化初始聚類中心的選擇，最后，基于Flink并行計(jì)算框架進(jìn)行實(shí)現(xiàn)。實(shí)驗(yàn)結(jié)果表明，相比原算法，本文算法在聚類準(zhǔn)確率和計(jì)算效率方面都有一定程度的提升，并且驗(yàn)證了在大規(guī)模數(shù)據(jù)背景下并行化實(shí)現(xiàn)的高效性。