基于Hadoop的K-Means算法的設計與實現

王立友 鄭海鵬

（淮南聯合大學計算機系 安徽淮南 232001）

在現實生活中，人們經常會將類似的事物或事情劃分為同一類。即所謂的“物以類聚、人以群分”。在計算機程序設計過程中，分類算法是一種監督學習方法。但是，在許多情況下，如果數據通過預處理滿足分類算法的要求，則無法滿足上述條件，特別是在處理大量數據時，成本非常大，您可以考慮使用K-Means聚類算法。K-Means算法屬于無監督學習方法[1]，作為較為經典的分類算法，在很多應用領域中都有其自身的優勢，因為它與分類進行比較，所以聚類不依賴于預定義的類和類標簽的訓練實例。

一、K-Means算法簡介

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

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

1)選擇初始化過程中中心點的個數（K值）：A1,A2,…,Ak；

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

1＜=j＜=k

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

4）重復2、3以上兩個步驟，直到迭代終止（即上一次和本次聚類的中心點不再發生變化）。

圖1 K-Means聚類示意圖

二、Hadoop平臺簡介

Hadoop是Apache Foundation開發的分布式系統基礎架構[3]。其核心是MapReduce和HDFS（分布式文件系統）。在現代社會中，只要和海量數據有關的應用領域都會出現Hadoop 的身影[4]。隨著Hadoop 版本的不斷發展，其自身的功能正在變得越來越強大，并且已經形成了相對完整的系統。如圖2所示。

圖2 hadoop平臺架構

（一）HDFS 分布式文件存儲系統。HDFS 代表Hadoop分布式文件系統，是一種具有高容錯性，高吞吐量和高可擴展性的分布式文件存儲系統。HDFS 在存儲數據時使用塊存儲。它將節點分為兩類：NameNode 和DataNode，其中NameNode是HDFS集群的主節點。DataNode是HDFS集群從節點，用來存儲真正的數據。每個數據塊可以在多個DataNode上存儲多個副本，以提高數據安全性。

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

（三）HBase數據庫。HBase是一個針對結構化數據的可伸縮、高性能的非關系型分布式動態模式數據庫，與傳統數據庫相比，HBase采用BigTable數據模型。在Hadoop平臺上運行時，HBase 中的表可用作MapReduce 任務的輸入和輸出，并可通過Java API檢索。

（四）Hive數據倉庫。Hive是數據倉庫管理文件。它提供完整的SQL 查詢功能，可將數據文件映射到單個數據表中。將其轉化為MapReduce任務進行運行。

三、基于Hadoop的K-means算法的實現

（一）K-Means算法實現過程解析。

1.使用全局變量在最后一次迭代后存儲質心。

2.使用Hadoop中的地圖框架（map）來計算每個質心和樣本之間的距離，并獲得最接近樣本的質心。

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

4.將先前質心和最新質心作比較，若發生了變化，就需要程序繼續運行，進行下一次迭代，否則將停止迭代，終止當前程序，輸出最終的聚類結果。

（二）算法實現過程需要解決的問題。本文的思路基本上是按照上面的步驟來做的，有以下幾個問題急需解決：

1.Hadoop沒有自定義全局變量，因此無法定義存儲質心的全局變量，因此需要另一種想法，即質心存儲在文件中。

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

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

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

SSE 方法的原理為：隨著聚類中心數量的增加，聚類將更加細化，每個聚類中心的聚合度將逐漸增加，因此誤差SSE將逐漸減小。當K小于實際簇數時，隨著K值的增加，簇中心的聚集程度將大大加快。加速誤差的平方和SSE減小的幅度。SSE 的下降趨勢急劇減少，然后趨于平緩，即SSE和K之間的關系是肘形。而對應肘部的K值為最佳。

為測試數據選擇最佳簇編號K值，具體實施過程為K=1-6，針對對每一個K值記下對應的SSE，制作K和SSE關系圖，并選擇與肘相對應的K 值作為最佳簇數。如3 所示：即K=4聚類結果最優。

圖3 K與SSE關系圖

（四）Hadoop實現K-Means算法的主要代碼。

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

public class Center{

protected static int k=4;//質心的個數（由手肘法確定K=4為最佳）

/**

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

*/

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.每次迭代從質心文件中讀取質心并返回字符串代碼

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中的聚類中心轉換為數組的形式，方便使用

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.求得成員距離最近的質心

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;

//計算相應維度上的值之和，存放在sum數組中

for(Text text : values) {

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

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

size ++;}//求新的質心

StringBuffer sb = new StringBuffer();

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

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

sb.append(","); }

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

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);

/**獲取自定義計數器的大小，如果它等于質心的大小，表示質心沒有改變，程序停止迭代*/

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; }}//如質心不一致，計數器加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");//初始質心文件

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

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

String centerString = center.loadInitCenter(kMeansPath);

Int index = 0;//初始化迭代次數

while(index < 100) {

Configuration conf = new Configuration();

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

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

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

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.重新加載質心

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);}}

四、結語

經在Hadoop 框架上進行測試，上述程序實現了簡單的K-Means 算法，并得出了較為理想的聚類效果。有關KMeans聚類算法在實際中的具體應用，筆者將在后續的實踐中繼續進行探究。