基于Hadoop的K-Means算法的設(shè)計(jì)與實(shí)現(xiàn)

王立友 鄭海鵬

（淮南聯(lián)合大學(xué)計(jì)算機(jī)系 安徽淮南 232001）

在現(xiàn)實(shí)生活中，人們經(jīng)常會(huì)將類似的事物或事情劃分為同一類。即所謂的“物以類聚、人以群分”。在計(jì)算機(jī)程序設(shè)計(jì)過(guò)程中，分類算法是一種監(jiān)督學(xué)習(xí)方法。但是，在許多情況下，如果數(shù)據(jù)通過(guò)預(yù)處理滿足分類算法的要求，則無(wú)法滿足上述條件，特別是在處理大量數(shù)據(jù)時(shí)，成本非常大，您可以考慮使用K-Means聚類算法。K-Means算法屬于無(wú)監(jiān)督學(xué)習(xí)方法[1]，作為較為經(jīng)典的分類算法，在很多應(yīng)用領(lǐng)域中都有其自身的優(yōu)勢(shì)，因?yàn)樗c分類進(jìn)行比較，所以聚類不依賴于預(yù)定義的類和類標(biāo)簽的訓(xùn)練實(shí)例。

一、K-Means算法簡(jiǎn)介

K-Means算法，也稱為K-Means方法或K均值法，具有良好的可擴(kuò)展性和快速收斂性[2]，是最廣泛使用的聚類算法。K均值算法是不斷改變聚類中心點(diǎn)的過(guò)程（聚類過(guò)程如圖1所示）。計(jì)算所有成員到每個(gè)質(zhì)心的距離，然后重新劃分其內(nèi)部集群成員。將對(duì)應(yīng)成員劃分給到其距離最小的那個(gè)質(zhì)心。K值表示簇的質(zhì)心數(shù)（即聚類中心的數(shù)量）；K-means可以自動(dòng)將對(duì)應(yīng)的成員分配給不同的類，但是不可能決定要分割哪些類。K必須是小于訓(xùn)練集中樣本數(shù)的正整數(shù)。

假設(shè)輸入成員樣本集合為C=X1,X2,…,Xn-1,Xn(含有n個(gè)成員樣本)，K-means算法具體的過(guò)程如下：

1)選擇初始化過(guò)程中中心點(diǎn)的個(gè)數(shù)（K值）：A1,A2,…,Ak；

2)計(jì)算成員樣本X1,X2,…,Xn-1,Xn到A1,A2,…,Ak的各自的距離；選取各個(gè)成員樣本到各個(gè)中心點(diǎn)的最小距離，將樣本成員劃分到最小距離的中心點(diǎn)中(如式1)；

1＜=j＜=k

3）重新為中心點(diǎn)Aj為隸屬該類別的所有樣本的均值；

4）重復(fù)2、3以上兩個(gè)步驟，直到迭代終止（即上一次和本次聚類的中心點(diǎn)不再發(fā)生變化）。

圖1 K-Means聚類示意圖

二、Hadoop平臺(tái)簡(jiǎn)介

Hadoop是Apache Foundation開發(fā)的分布式系統(tǒng)基礎(chǔ)架構(gòu)[3]。其核心是MapReduce和HDFS（分布式文件系統(tǒng)）。在現(xiàn)代社會(huì)中，只要和海量數(shù)據(jù)有關(guān)的應(yīng)用領(lǐng)域都會(huì)出現(xiàn)Hadoop 的身影[4]。隨著Hadoop 版本的不斷發(fā)展，其自身的功能正在變得越來(lái)越強(qiáng)大，并且已經(jīng)形成了相對(duì)完整的系統(tǒng)。如圖2所示。

圖2 hadoop平臺(tái)架構(gòu)

（一）HDFS 分布式文件存儲(chǔ)系統(tǒng)。HDFS 代表Hadoop分布式文件系統(tǒng)，是一種具有高容錯(cuò)性，高吞吐量和高可擴(kuò)展性的分布式文件存儲(chǔ)系統(tǒng)。HDFS 在存儲(chǔ)數(shù)據(jù)時(shí)使用塊存儲(chǔ)。它將節(jié)點(diǎn)分為兩類：NameNode 和DataNode，其中NameNode是HDFS集群的主節(jié)點(diǎn)。DataNode是HDFS集群從節(jié)點(diǎn)，用來(lái)存儲(chǔ)真正的數(shù)據(jù)。每個(gè)數(shù)據(jù)塊可以在多個(gè)DataNode上存儲(chǔ)多個(gè)副本，以提高數(shù)據(jù)安全性。

（二）MapReduce 框架。MapReduce 是一個(gè)開源并行計(jì)算框架，主要由兩個(gè)函數(shù)組成：Map（）和Reduce（）。它可以將大型任務(wù)分解為相對(duì)相反的多個(gè)子任務(wù)，并且可以并行處理。Map（）函數(shù)負(fù)責(zé)指定數(shù)據(jù)集上的相反元素，并生成一系列[key，value]作為中間結(jié)果[5]。而Reduce（）函數(shù)則對(duì)中間結(jié)果中相同Key的所有“value”進(jìn)行規(guī)約，并得到最終結(jié)果。

（三）HBase數(shù)據(jù)庫(kù)。HBase是一個(gè)針對(duì)結(jié)構(gòu)化數(shù)據(jù)的可伸縮、高性能的非關(guān)系型分布式動(dòng)態(tài)模式數(shù)據(jù)庫(kù)，與傳統(tǒng)數(shù)據(jù)庫(kù)相比，HBase采用BigTable數(shù)據(jù)模型。在Hadoop平臺(tái)上運(yùn)行時(shí)，HBase 中的表可用作MapReduce 任務(wù)的輸入和輸出，并可通過(guò)Java API檢索。

（四）Hive數(shù)據(jù)倉(cāng)庫(kù)。Hive是數(shù)據(jù)倉(cāng)庫(kù)管理文件。它提供完整的SQL 查詢功能，可將數(shù)據(jù)文件映射到單個(gè)數(shù)據(jù)表中。將其轉(zhuǎn)化為MapReduce任務(wù)進(jìn)行運(yùn)行。

三、基于Hadoop的K-means算法的實(shí)現(xiàn)

（一）K-Means算法實(shí)現(xiàn)過(guò)程解析。

1.使用全局變量在最后一次迭代后存儲(chǔ)質(zhì)心。

2.使用Hadoop中的地圖框架（map）來(lái)計(jì)算每個(gè)質(zhì)心和樣本之間的距離，并獲得最接近樣本的質(zhì)心。

3.在Hadoop 中使用reduce 框架，輸入鍵（Key）是質(zhì)心，值（Value）是其他樣本。

4.將先前質(zhì)心和最新質(zhì)心作比較，若發(fā)生了變化，就需要程序繼續(xù)運(yùn)行，進(jìn)行下一次迭代，否則將停止迭代，終止當(dāng)前程序，輸出最終的聚類結(jié)果。

（二）算法實(shí)現(xiàn)過(guò)程需要解決的問(wèn)題。本文的思路基本上是按照上面的步驟來(lái)做的，有以下幾個(gè)問(wèn)題急需解決：

1.Hadoop沒(méi)有自定義全局變量，因此無(wú)法定義存儲(chǔ)質(zhì)心的全局變量，因此需要另一種想法，即質(zhì)心存儲(chǔ)在文件中。

2.存儲(chǔ)質(zhì)心的文件讀取。我們選擇在主main函數(shù)中讀取質(zhì)心，然后將質(zhì)心設(shè)置保存到配置中。configuration 在map和reduce均可讀

3.如何比較質(zhì)心的變化，你可以在主要的主函數(shù)中進(jìn)行比較，讀取質(zhì)心的文件和最新的質(zhì)心并進(jìn)行比較。這個(gè)方法是可以實(shí)現(xiàn)的，我們使用自定義計(jì)數(shù)器，計(jì)數(shù)器是一個(gè)全局變量，可在地圖中讀寫并減少，在上面的想法中，我們看到reduce具有最后一次迭代的質(zhì)心和計(jì)算的質(zhì)心，因此完全可以直接在reduce中進(jìn)行比較。如果發(fā)生變化，counter加1。

（三）基于手肘法的K-Means 算法中最佳K 值的確定。本文選取手肘法用來(lái)確定最佳K值[6]。手肘法的核心指標(biāo)是平方誤差的總和：SSE（如式2所示），其中，Ci是第i個(gè)聚類中心，p是聚類中心的元素，mi是Ci的質(zhì)心（Ci中所有元素的平均值），SSE則可以表示聚類的質(zhì)量。

SSE 方法的原理為：隨著聚類中心數(shù)量的增加，聚類將更加細(xì)化，每個(gè)聚類中心的聚合度將逐漸增加，因此誤差SSE將逐漸減小。當(dāng)K小于實(shí)際簇?cái)?shù)時(shí)，隨著K值的增加，簇中心的聚集程度將大大加快。加速誤差的平方和SSE減小的幅度。SSE 的下降趨勢(shì)急劇減少，然后趨于平緩，即SSE和K之間的關(guān)系是肘形。而對(duì)應(yīng)肘部的K值為最佳。

為測(cè)試數(shù)據(jù)選擇最佳簇編號(hào)K值，具體實(shí)施過(guò)程為K=1-6，針對(duì)對(duì)每一個(gè)K值記下對(duì)應(yīng)的SSE，制作K和SSE關(guān)系圖，并選擇與肘相對(duì)應(yīng)的K 值作為最佳簇?cái)?shù)。如3 所示：即K=4聚類結(jié)果最優(yōu)。

圖3 K與SSE關(guān)系圖

（四）Hadoop實(shí)現(xiàn)K-Means算法的主要代碼。

1.定義Center類、初始化質(zhì)心.首先定義一個(gè)Center類，它主要存儲(chǔ)質(zhì)心數(shù)量K值，以及兩個(gè)從HDFS讀取質(zhì)心文件的方法。一個(gè)用于讀取初始質(zhì)心，一個(gè)用于在每次迭代后讀取質(zhì)心文件夾。這個(gè)是在文件夾中的，代碼如下：

public class Center{

protected static int k=4;//質(zhì)心的個(gè)數(shù)（由手肘法確定K=4為最佳）

/**

從初始質(zhì)心文件加載質(zhì)心并返回字符串，由質(zhì)心之間的制表符分隔

*/

public String loadInitCenter(Path path) throws IOException {

StringBuffer sb = new StringBuffer();

Configuration conf = new Configuration();

FileSystem hdfs = FileSystem.get(conf);

FSDataInputStream dis = hdfs.open(path);

LineReader in = new LineReader(dis,conf);

Text line = new Text();

while(in.readLine(line) ＞0) {

sb.append(line.toString().trim());

sb.append(" "); }

return sb.toString().trim();}

2.每次迭代從質(zhì)心文件中讀取質(zhì)心并返回字符串代碼

public String loadCenter(Path path) throws IOException{

StringBuffer sb = new StringBuffer();

Configuration conf = new Configuration();

FileSystem hdfs = FileSystem.get(conf);

FileStatus[]files = hdfs.listStatus(path);

for(int i = 0; i < files.length; i++) {

Path filePath = files[i].getPath();

if(!filePath.getName().contains("part")) continue;

FSDataInputStream dis = hdfs.open(filePath);

LineReader in = new LineReader(dis,conf);

Text line = new Text();

while(in.readLine(line) ＞0) {

sb.append(line.toString().trim());

sb.append(" ");}}

return sb.toString().trim();}}

public class KmeansMR {

private static String FLAG = "KCLUSTER";

public static class TokenizerMapper extends Mapper

{ double[][]centers = new double[Center.k][];

String[]centerstrArray = null;

@Override

public void setup(Context context) {

//將放在context中的聚類中心轉(zhuǎn)換為數(shù)組的形式，方便使用

String kmeansS = context.getConfiguration().get(FLAG);

centerstrArray = kmeansS.split(" ");

for(int i = 0; i < centerstrArray.length; i++) {

String[]segs = centerstrArray[i].split(",");

centers[i]= new double[segs.length];

for(int j = 0; j < segs.length; j++) {

centers[i][j]= Double.parseDouble(segs[j]);}}}

public void map(Object key,Text value,Context context) throws IOException,InterruptedException {

String line = value.toString();

String[]segs = line.split(",");

double[]sample = new double[segs.length];

for(int i = 0; i < segs.length; i++) {

sample[i]= Float.parseFloat(segs[i]);}

3.求得成員距離最近的質(zhì)心

double min = Double.MAX_VALUE;

int index = 0;

for(int i = 0; i < centers.length; i++) {

double dis = distance(centers[i],sample);

if(dis < min) { min = dis;index = i; } }

context.write(new Text(centerstrArray[index]),new Text(line));

} }

public static class IntSumReducer

extends Reducer

Counter counter = null;

public void reduce(Text key,Iterable

double[]sum = new double[Center.k]; int size = 0;

//計(jì)算相應(yīng)維度上的值之和，存放在sum數(shù)組中

for(Text text : values) {

String[]segs = text.toString().split(",");

for( int i = 0; i < segs.length; i++) {sum[i]+= Double.parseDouble(segs[i]);}

size ++;}//求新的質(zhì)心

StringBuffer sb = new StringBuffer();

for(int i = 0; i < sum.length; i++) {

sum[i]/= size; sb.append(sum[i]);

sb.append(","); }

4.判斷新的質(zhì)心跟前一次迭代的質(zhì)心是否一致

job.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(NullWritable.class);

job.setOutputValueClass(Text.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(Text.class);

FileInputFormat.addInputPath(job,samplePath);

FileOutputFormat.setOutputPath(job,kMeansPath);

job.waitForCompletion(true);

/**獲取自定義計(jì)數(shù)器的大小，如果它等于質(zhì)心的大小，表示質(zhì)心沒(méi)有改變，程序停止迭代*/

long counter = job.getCounters().getGroup("myCounter").findCounter("kmenasCounter").getValue();

if(counter == Center.k) System.exit(0);

boolean flag = true;

String[]centerStrArray = key.toString().split(",");

for(int i = 0; i < centerStrArray.length; i++) {

if(Math.abs(Double.parseDouble(centerStrArray[i]) - sum[i]) ＞0.00000000001)

{ flag = false; break; }}//如質(zhì)心不一致，計(jì)數(shù)器加1

if(flag) {counter = context.getCounter("myCounter","kmenasCounter");

counter.increment(1l); }

context.write(null,new Text(sb.toString()));}}

public static void main(String[]args) throws Exception {

Path kMeansPath = new Path("/mid-data/k-means/k-Means");//初始質(zhì)心文件

Path samplePath = new Path("/mid-data/k-means/sample");//樣本文件

Center center = new Center();//加載聚類中心文件

String centerString = center.loadInitCenter(kMeansPath);

Int index = 0;//初始化迭代次數(shù)

while(index < 100) {

Configuration conf = new Configuration();

conf.set(FLAG,centerString);////將集群中心字符串放入配置中

kMeansPath = new Path("mid-data/k-means/k-Means" +index);//此迭代的輸出路徑也是下一個(gè)質(zhì)心的讀取路徑

/**確定輸出路徑是否存在*/

FileSystem hdfs = FileSystem.get(conf);

if(hdfs.exists(kMeansPath)) hdfs.delete(kMeansPath);

Job job = new Job(conf,"kmeans" + index);

job.setJarByClass(KmeansMR.class);

job.setMapperClass(TokenizerMapper.class);

5.重新加載質(zhì)心

center = new Center();centerString = center.loadCenter(kMeansPath);

index ++;}

System.exit(0);}

public static double distance(double[]a,double[]b) {

if(a == null || b == null || a.length != b.length) return Double.MAX_VALUE;

double dis = 0;

for(int i = 0; i < a.length; i++) { dis += Math.pow(a[i]-b[i],2);}

return Math.sqrt(dis);}}

四、結(jié)語(yǔ)

經(jīng)在Hadoop 框架上進(jìn)行測(cè)試，上述程序?qū)崿F(xiàn)了簡(jiǎn)單的K-Means 算法，并得出了較為理想的聚類效果。有關(guān)KMeans聚類算法在實(shí)際中的具體應(yīng)用，筆者將在后續(xù)的實(shí)踐中繼續(xù)進(jìn)行探究。