接着上篇,继续分析代码。下面就到了MR的循环了,这里MR应该算是比较好理解的,重点是退出循环的条件设置,即如何判断前后两次中心点误差小于给定阈值。
首先,while循环:
while (iteration <= numIterations) {
conf.set(PRIOR_PATH_KEY, priorPath.toString());
String jobName = "Cluster Iterator running iteration " + iteration + " over priorPath: " + priorPath;
System.out.println(jobName);
Job job = new Job(conf, jobName);
job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(ClusterWritable.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(ClusterWritable.class);
job.setInputFormatClass(SequenceFileInputFormat.class);
job.setOutputFormatClass(SequenceFileOutputFormat.class);
job.setMapperClass(CIMapper.class);
job.setReducerClass(CIReducer.class);
FileInputFormat.addInputPath(job, inPath);
clustersOut = new Path(outPath, Cluster.CLUSTERS_DIR + iteration);
priorPath = clustersOut;
FileOutputFormat.setOutputPath(job, clustersOut);
job.setJarByClass(ClusterIterator.class);
if (!job.waitForCompletion(true)) {
throw new InterruptedException("Cluster Iteration " + iteration + " failed processing " + priorPath);
}
ClusterClassifier.writePolicy(policy, clustersOut);
FileSystem fs = FileSystem.get(outPath.toUri(), conf);
iteration++;
if (isConverged(clustersOut, conf, fs)) {
break;
}
}
这个循环可以看出每个MR的输入都是一样的,输出为outPath+"/clusters-"+iteration ,然后每次MR后就会把同一个policy写入输出里面,循环次数加1,然后判断是否退出循环即isConverged(clustersOut,conf,fs)方法;下面看仿造版的MR。首先看Mapper:
package mahout.fansy.kmeans;
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.mahout.clustering.Cluster;
import org.apache.mahout.clustering.classify.ClusterClassifier;
import org.apache.mahout.clustering.iterator.ClusterIterator;
import org.apache.mahout.clustering.iterator.ClusterWritable;
import org.apache.mahout.clustering.iterator.ClusteringPolicy;
import org.apache.mahout.common.iterator.sequencefile.PathFilters;
import org.apache.mahout.common.iterator.sequencefile.PathType;
import org.apache.mahout.common.iterator.sequencefile.SequenceFileDirValueIterable;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.mahout.math.Vector.Element;
import com.google.common.collect.Lists;
public class TestCIMapper {
/**
* @param args
*/
private static ClusterClassifier classifier;
private static ClusteringPolicy policy;
public static void main(String[] args) throws IOException {
setup();
map();
cleanup();
}
/**
* 仿造setup函数
* @throws IOException
*/
public static void setup() throws IOException{
Configuration conf=new Configuration();
conf.set("mapred.job.tracker", "hadoop:9001"); // 这句是否可以去掉?
String priorClustersPath ="hdfs://hadoop:9000/user/hadoop/out/kmeans-output/clusters-0";
classifier = new ClusterClassifier();
classifier.readFromSeqFiles(conf, new Path(priorClustersPath));
policy = classifier.getPolicy();
policy.update(classifier);
}
/**
* 仿造map函数
*/
public static void map(){
List<VectorWritable> vList=getInputData();
for(VectorWritable value: vList){
Vector probabilities = classifier.classify(value.get());
Vector selections = policy.select(probabilities);
for (Iterator<Element> it = selections.iterateNonZero(); it.hasNext();) {
Element el = it.next();
classifier.train(el.index(), value.get(), el.get());
}
}
}
/**
* 仿造cleanup函数
*/
public static void cleanup(){
List<Cluster> clusters = classifier.getModels();
ClusterWritable cw = new ClusterWritable();
for (int index = 0; index < clusters.size(); index++) {
cw.setValue(clusters.get(index));
System.out.println("index:"+index+",cw :"+ cw.getValue().getCenter() );
}
}
/**
* 获得输入数据
* @return
*/
public static List<VectorWritable> getInputData(){
String input="hdfs://hadoop:9000/user/hadoop/out/kmeans-in-transform/part-r-00000";
Path path=new Path(input);
Configuration conf=new Configuration();
List<VectorWritable> vList=Lists.newArrayList();
for (VectorWritable cw : new SequenceFileDirValueIterable<VectorWritable>(path, PathType.LIST,
PathFilters.logsCRCFilter(), conf)) {
vList.add(cw);
}
return vList;
}
}
上面的代码中的setup函数函数就是把中心点和阈值读入变量而已,比如classifier的变量如下:
其实感觉好像在前面代码的处理中没有必要一个中心点使用一个文件存储吧?这里把这些文件又都读入一个变量了,还要那么多文件干嘛呢?或许这个是某个大神的得意之作,而我还没有发现其精妙之处?还有就是前面的policy变量也应该没有必要存入到文件吧,这里的classifier变量里面都有policy变量了。
下面是map函数,在map函数之前有个getInputData方法用于获得输入数据,把输入数据存入一个变量中。然后在map中foreach读出来。
在map函数中,其实只有这三句有用:
Vector probabilities = classifier.classify(value.get());
Vector selections = policy.select(probabilities);
classifier.train(el.index(), value.get(), el.get());
第一行,没有设置到关于classifier的代码,第二行也是,第三行虽然设置了classifier中的models属性,但是没有改变其elementData中的center属性,而是改S0,S1之类的,如果这样没有改动的话,那在cleanup里面又是直接输出这个classifier的,那么就等于是没有改到中心点向量,那是怎么更新的呢?额 太困了,下次继续。。。
分享,快乐,成长
转载请注明出处:http://blog.csdn.net/fansy1990
分享到:
相关推荐
mahoutAlgorithms源码分析 mahout代码解析
mahout KMeansDriver测试相关jar包,有需要的同学可以下载来试试。把这个包放入hadoop/lib下面然后就可以在eclipse里面运行调试了。
Mahout是一个Java的机器学习库。Mahout的完整源代码,基于maven,可以轻易导入工程中
mahout,朴素贝叶斯分类,中文分词,mahout,朴素贝叶斯分类,中文分词,
mahout-distribution-0.5-src.zip mahout 源码包
Mahout 是 Apache Software Foundation(ASF) 旗下的一个开源项目,提供一些可扩展的机器学习领域经典算法的实现,旨在帮助开发人员更加方便快捷地创建智能应用程序。Mahout包含许多实现,包括聚类、分类、推荐过滤...
mahout 0.7 src, mahout 源码包, hadoop 机器学习子项目 mahout 源码包
mahout0.9的源码,支持hadoop2,需要自行使用mvn编译。mvn编译使用命令: mvn clean install -Dhadoop2 -Dhadoop.2.version=2.2.0 -DskipTests
mahout in action中的example codes进行maven编译时由于maven相关jar包的URL的重定位,故无法进行有效编译,需要下载相关jar包进行手动加载!
svd算法的工具类,直接调用出结果,调用及设置方式参考http://blog.csdn.net/fansy1990 <mahout源码分析之DistributedLanczosSolver(七)>
mahout实战 源码 mahout实战 配套 mahout-distribution-0.5.tar.gz 版本
该资源是mahout in action 中的源码,适用于自学,可在github下载:https://github.com/tdunning/MiA
【甘道夫】通过Mahout构建贝叶斯文本分类器案例详解 -- 配套源码
Thank you for requesting the download for Apache Mahout Cookbook. Please click the following link to download the code:
Mahout in Action 源码,结合Mahout in Action 学习数据挖掘,比较容易理解
基于Mahout协同过滤实现图书推荐系统_书籍推荐系统源码+项目说明.zip 基于协同过滤的书籍推荐系统,图书推荐系统 最新版本,在原先手动计算皮尔逊相似度和评分矩阵的基础上添加了Mahout实现的协同过滤推荐算法. ...
mahout0.11版本,源码,可修改源码并自己编译,使用java语言编写,maven编译
Mahout:整体框架,实现了协同过滤 Deeplearning4j,构建VSM Jieba:分词,关键词提取 HanLP:分词,关键词提取 Spring Boot:提供API、ORM 关键实现 基于用户的协同过滤 直接调用Mahout相关接口即可 选择不同...
基于Java+Mahout的协同过滤推荐算法图书推荐系统源码+项目说明.zip 基于协同过滤的书籍推荐系统,图书推荐系统 最新版本,在原先手动计算皮尔逊相似度和评分矩阵的基础上添加了Mahout实现的协同过滤推荐算法。 ...