mapreduce中怎么实现K-M类聚，针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。
首先是map
public static class KMmap extends Mapper<LongWritable, Text, IntWritable, Text>{
//中心集合
//这里的聚簇集合是自己设定的 centersPath就是集合在hdfs中存放的路径
ArrayList<ArrayList<Double>> centers = null;
//用k个中心
int k = 0;
//读取中心
protected void setup(Context context)throws IOException, InterruptedException {
//getCentersFromHDFS方法就是传入一个Path，得到一个ArrayList<ArrayList<Double>>集合
centers = Utils.getCentersFromHDFS(context.getConfiguration().get("centersPath"),false);
k = centers.size();
}
/**
* 1.每次读取一条要分类的条记录与中心做对比，归类到对应的中心
* 2.以中心ID为key，中心包含的记录为value输出(例如： 1 0.2 。 1为聚类中心的ID，0.2为靠近聚类中心的某个值)
*/
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
ArrayList<Double> fileds = Utils.textToArray(value);
//textToArray方法将map进来的一行value根据“,”分割后转化为ArrayList<Double>的集合
int sizeOfFileds = fileds.size();
double minDistance = 99999999;
int centerIndex = 0;
//依次取出k个中心点与当前读取的记录做计算
for(int i=0;i<k;i++){
double currentDistance = 0;
for(int j=0;j<sizeOfFileds;j++){
//centers中存放的是中心点集合
//fileds中存放的是被分析数据的每行的数据集合
double centerPoint = Math.abs(centers.get(i).get(j));
double filed = Math.abs(fileds.get(j));
//根据类聚的公式先对[(h-k)/(h+k)]²累加
currentDistance += Math.pow((centerPoint - filed) / (centerPoint + filed), 2);
}
//循环找出距离该记录最接近的中心点的ID
if(currentDistance<minDistance){
minDistance = currentDistance;
centerIndex = i;
}
}
//以中心点为Key 将记录原样输出
context.write(new IntWritable(centerIndex+1), value);
}
}reduce
//利用reduce的归并功能以中心为Key将记录归并到一起
public static class KMreduce extends Reducer<IntWritable, Text, Text, Text>{
/**
* 1.Key为聚类中心的ID value为该中心的记录集合
* 2.计数所有记录元素的平均值，求出新的中心
*/
protected void reduce(IntWritable key, Iterable<Text> values,
Context context)throws IOException, InterruptedException {
ArrayList<ArrayList<Double>> filedsList = new ArrayList<ArrayList<Double>>();
//依次读取记录集，每行为一个ArrayList<Double>
for(Iterator<Text> it = values.iterator();it.hasNext();){
ArrayList<Double> tempList = Utils.textToArray(it.next());
filedsList.add(tempList);
}
//计算新的中心
//每行的元素个数
int filedSize = filedsList.get(0).size();
double[] avg = new double[filedSize];
for(int i=0;i<filedSize;i++){
//求每列的平均值
double sum = 0;
int size = filedsList.size();
for(int j=0;j<size;j++){
sum += filedsList.get(j).get(i);
}
avg[i] = sum/size;
}
context.write(new Text(""), new Text((Arrays.toString(avg).replace("[", "").replace("]", ""))));
}
}最后是其中所用到的util类，主要是提供一些读取文件和操作字符串的方法
public class Utils {
//读取中心文件的数据
public static ArrayList<ArrayList<Double>> getCentersFromHDFS(String centersPath,boolean isDirectory)
throws IOException{
ArrayList<ArrayList<Double>> result = new ArrayList<ArrayList<Double>>();
Path path = new Path(centersPath);
Configuration conf = new Configuration();
FileSystem fileSystem = path.getFileSystem(conf);
if(isDirectory){
FileStatus[] listFile = fileSystem.listStatus(path);
for (int i = 0; i < listFile.length; i++) {
result.addAll(getCentersFromHDFS(listFile[i].getPath().toString(),false));
}
return result;
}
FSDataInputStream fsis = fileSystem.open(path);
LineReader lineReader = new LineReader(fsis, conf);
Text line = new Text();
while(lineReader.readLine(line) > 0){
ArrayList<Double> tempList = textToArray(line);
result.add(tempList);
}
lineReader.close();
return result;
}
//删掉文件
public static void deletePath(String pathStr) throws IOException{
Configuration conf = new Configuration();
Path path = new Path(pathStr);
FileSystem hdfs = path.getFileSystem(conf);
hdfs.delete(path ,true);
}
public static ArrayList<Double> textToArray(Text text){
ArrayList<Double> list = new ArrayList<Double>();
String[] fileds = text.toString().split("\t");
for(int i=0;i<fileds.length;i++){
list.add(Double.parseDouble(fileds[i]));
}
return list;
}
public static boolean compareCenters(String centerPath,String newPath) throws IOException{
List<ArrayList<Double>> oldCenters = Utils.getCentersFromHDFS(centerPath,false);
List<ArrayList<Double>> newCenters = Utils.getCentersFromHDFS(newPath,true);
int size = oldCenters.size();
int fildSize = oldCenters.get(0).size();
double distance = 0;
for(int i=0;i<size;i++){
for(int j=0;j<fildSize;j++){
double t1 = Math.abs(oldCenters.get(i).get(j));
double t2 = Math.abs(newCenters.get(i).get(j));
distance += Math.pow((t1 - t2) / (t1 + t2), 2);
}
}
if(distance == 0.0){
//删掉新的中心文件以便最后依次归类输出
Utils.deletePath(newPath);
return true;
}else{
//先清空中心文件，将新的中心文件复制到中心文件中，再删掉中心文件
Configuration conf = new Configuration();
Path outPath = new Path(centerPath);
FileSystem fileSystem = outPath.getFileSystem(conf);
FSDataOutputStream overWrite = fileSystem.create(outPath,true);
overWrite.writeChars("");
overWrite.close();
Path inPath = new Path(newPath);
FileStatus[] listFiles = fileSystem.listStatus(inPath);
for (int i = 0; i < listFiles.length; i++) {
FSDataOutputStream out = fileSystem.create(outPath);
FSDataInputStream in = fileSystem.open(listFiles[i].getPath());
IOUtils.copyBytes(in, out, 4096, true);
}
//删掉新的中心文件以便第二次任务运行输出
Utils.deletePath(newPath);
}
return false;
}
}关于mapreduce中怎么实现K-M类聚问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注亿速云行业资讯频道了解更多相关知识。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
