Hadoop Web项目--Friend Find系统

项目使用软件：Myeclipse10.0，JDK1.7，Hadoop2.6，MySQL5.6。EasyUI1.3.6。jQuery2.0，Spring4.1.3。

Hibernate4.3.1，struts2.3.1。Tomcat7 。Maven3.2.1。

项目下载地址：https://github.com/fansy1990/friend_find ，项目部署參考：http://blog.csdn.net/fansy1990/article/details/46481409 。

Hadoop Web项目--Friend Find系统

1. 项目介绍

        Friend Find系统是一个寻找相似用户的系统。

用户填写本身的信息后就可以在本系统内找到和本身志同道合的朋友。本系统使用的是在http://stackoverflow.com/站点上的用户数据。Stack Overflow是一个程序设计领域的问答站点，隶属Stack Exchange Network。站点赞成注冊用户提出或回答问题。还赞成对已有问题或答案加分、扣分或进行改动，条件是用户达到必定的“声望值”。“声望值”就是用户进行站点交互时能获取的分数。当声望值达到某个程度时，用户的权限就会添加，比方声望值超过50点就可以评论答案。当用户的声望值达到某个阶段时，站点还会给用户颁发贡献徽章。以此来激励用户对站点作出贡献。该项目创建在如下的假设基础上，假设用户对于一个领域问题的“态度”就可以反映出该用户的价值取向，并依据此价值取向来对用户进行聚类分组。

这里的态度可以使用几个指标属性来评判，在本系统中原始数据（即用户信息数据）包含的属性有多个，从中挑选出最能符合用户观点的属性。做为该用户的“态度”进行分析。这里挑选的属性是：reputation，upVotes。downVotes，views。即便用这4个属性来对用户进行聚类。同一时候，这里使用MR实现的Clustering by fast search and find of density peaks聚类算法，这里的实现和http://blog.csdn.net/fansy1990/article/details/46364697这里的实现原始是不一样的。

2. 项目执行

2.1 准备

1. 下载project。參考上面的链接 https://github.com/fansy1990/friend_find，并參考 http://blog.csdn.net/fansy1990/article/details/46481409把它部署上去；

 1） 注意依据数据库的配置，在mysql数据库中新建一个friend数据库；

 2）直接执行部署project，就能够在数据库中本身主动创建相应的表，包含：hconstants、loginuser、userdata、usergroup，当中loginuser是用户登陆表。会本身主动初始化（默认有两个用户admin/admin、test/test），hconstants是云平台參数数据表、userdata存储原始用户数据、usergroup存储聚类分群后每个用户的组别。

2. 部署云平台Hadoop2.6（伪分布式或者全然分布式都可以，本项目測试使用伪分布式），同一时候需要注意：设置云平台系统linux的时间和执行tomcat的机器的时间同样，因为在云平台任务监控的时候使用了时间做为监控中止的信号（详细可以參考后面）。

3. 使用MyEclipse的export功能把所有源代码打包，而后把打包后的jar文件复制到hadoop集群的$HADOOP_HOME/share/hadoop/mapreduce/文件夹如下。

2.2 执行

1. 初始化相应的表

初始化集群配置表hconstants

訪问系统首页：http://localhost/friend_find (这里部署的tomcat默认使用80端口，同一时候web部署的名称为friend_find),就能够看到如下的页面(系统首页)：

点击登陆，就能够看到系统介绍。

点击初始化表，依次选择相应的表。就能够完毕初始化

点击Hadoop集群配置表，查看数据：

这里初始化使用的是lz的虚拟机的配置。因此需要改动为本身的集群配置，点击某一行数据。在toolbar里就能够选择改动或保存等。

2. 系统原始文件：

系统原始文件在project的：

3. 项目实现流程

项目实现的流程依照系统首页左边导航栏的顺序从上到下执行。完毕数据挖掘的各个步骤。

3.1 数据探索

下载原始数据ask_ubuntu_users.xml 文件，打开，可以看到：

原始数据一共同拥有19550条记录，去除第一、二、最后一行外其它都是用户数据（第3行不是用户数据，是该站点的描写叙述）；

用户数据需要使用一个主键来惟一标示该用户。这里不是选择Id。而是使用EmailHash（这里假设每个EmailHash一样的帐号其是同一我的）。

使用上面的假设后，对原始数据进行分析（这里是所有导入到数据库后发现的），发现EmailHash是有反复记录的。因此这里需要对数据进行预处理--去重。

3.2 数据预处理

1. 数据去重

数据去重採用云平台Hadoop进行处理，首先把ask_ubuntu_users.xml文件上传到云平台，接着执行MR任务进行过滤。

2. 数据序列化

因为计算用户向量两两之间的距离的MR任务使用的是序列化的文件，因此这里需要对数据进行序列化处理；

3.3 建模

建模即便用高速聚类算法来对原始数据进行聚类，主要包含如下几个步骤：

1. 计算用户向量两两之间的距离；

2. 依据距离求解每个用户向量的局部密度。

3. 依据1.和2.的结果求解每个用户向量的最小距离；

4. 依据2，3的结果画出决策图，并推断聚类中心的局部密度和最小距离的阈值。

5. 依据局部密度和最小距离阈值来寻找聚类中心向量；

6. 依据聚类中心向量来进行分类；

3.4 推荐

建模后的结果即可以获得聚类中心向量以及每个分群的百分比，同一时候依据分类的结果来对用户进行组内推荐。

项目流程图例如如下：

4. 项目功能及实现原理

项目功能主要包含如下：

4.1 数据库表维护

数据库表维护主要包含：数据库表初始化，即用户登陆表和Hadoop集群配置表的初始化。数据库表增删改查查看：即用户登陆表、用户数据表、Hadoop集群配置表的增删改查。

数据库表增删改查使用同一个DBService类来进行处理。（这里的DAO使用的是通用的）假设针对每个表都创建一个DAO，那么代码就很是臃肿，因此这里把这些数据库表都是实现一个接口ObjectInterface，该接口使用一个Map来实例化各个对象。

public interface ObjectInterface {
	/**
	 * 不用每个表都创建一个方法，这里依据表名本身主动装配
	 * @param map
	 * @return
	 */
	public  Object setObjectByMap(Map<String,Object> map);
}

在进行保存的时候，直接使用前台传入的表名和json字符串进行更新就能够

/**
	 * 更新或者插入表
	 * 不用每个表都创建一个方法，这里依据表名本身主动装配
	 * @param tableName
	 * @param json
	 * @return
	 */
	public boolean updateOrSave(String tableName,String json){
		try{
			// 依据表名得到实体类。并赋值
			Object o = Utils.getEntity(Utils.getEntityPackages(tableName),json);
			baseDao.saveOrUpdate(o);
			log.info("保存表{}。",new Object[]{tableName});
		}catch(Exception e){
			
			e.printStackTrace();
			return false;
		}
		return true;
	}

/**
	 * 依据类名得到实体类
	 * @param tableName
	 * @param json
	 * @return
	 * @throws ClassNotFoundException 
	 * @throws IllegalAccessException 
	 * @throws InstantiationException 
	 * @throws IOException 
	 * @throws JsonMappingException 
	 * @throws JsonParseException 
	 */
	@SuppressWarnings("unchecked")
	public static Object getEntity(String tableName, String json) throws ClassNotFoundException, InstantiationException, IllegalAccessException, JsonParseException, JsonMappingException, IOException {
		Class<?
> cl = Class.forName(tableName);
		ObjectInterface o = (ObjectInterface)cl.newInstance();
		Map<String,Object> map = new HashMap<String,Object>();
		ObjectMapper mapper = new ObjectMapper();
		try {
			//convert JSON string to Map
			map = mapper.readValue(json, Map.class);
			return o.setObjectByMap(map);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return null;
	}

4.2 数据预处理

数据预处理包含文件上传、文件去重、文件下载、数据入库、DB过滤到HDFS、距离计算、最佳DC。

1. 文件上传

文件上传便是把文件从本地上传到HDFS，例如如下界面：

这里上传的便是ask_ubuntu_users.xml 所有数据文件。

上传直接使用FileSystem的静态方法下载，例如如下代码（）：

fs.copyFromLocalFile(src, dst);

上传成功就能够显示操做成功，这里使用aJax异步提交：

// =====uploadId,数据上传button绑定 click方法
	$('#uploadId').bind('click', function(){
		var input_i=$('#localFileId').val();
		// 弹出进度框
		popupProgressbar('数据上传','数据上传中...',1000);
		// ajax 异步提交任务
		callByAJax('cloud/cloud_upload.action',{input:input_i});
	});

当中调用aJax使用一个封装的方法。之后都可以调用，例如如下：

// 调用ajax异步提交
// 任务返回成功。则提示成功。不然提示失败的信息
function callByAJax(url,data_){
	$.ajax({
		url : url,
		data: data_,
		async:true,
		dataType:"json",
		context : document.body,
		success : function(data) {
//			$.messager.progress('close');
			closeProgressbar();
			console.info("data.flag:"+data.flag);
			var retMsg;
			if("true"==data.flag){
				retMsg='操做成功。';
			}else{
				retMsg='操做失败。失败缘由：'+data.msg;
			}
			$.messager.show({
				title : '提示',
				msg : retMsg
			});
			
			if("true"==data.flag&&"true"==data.monitor){// 加入监控页面
				// 使用单独Tab的方式
				layout_center_addTabFun({
					title : 'MR算法监控',
					closable : true,
					// iconCls : node.iconCls,
					href : 'cluster/monitor_one.jsp'
				});
			}
			
		}
	});
}

后台返回的是json数据。并且这里为了和云平台监控任务兼容（考虑通用性），这里还加入了一个打开监控的代码。

2. 文件去重

在导航栏选择文件去重，就能够看到如下的界面：

点击去重就能够提交任务到云平台，并且会打开MR的监控，例如如下图：

在点击”去重“按钮时，会启动一个后台线程Thread：

/**
	 * 去重任务提交
	 */
	public void deduplicate(){
		Map<String ,Object> map = new HashMap<String,Object>();
		try{
			HUtils.setJobStartTime(System.currentTimeMillis()-10000);
			HUtils.JOBNUM=1;
			new Thread(new Deduplicate(input,output)).start();
			map.put("flag", "true");
			map.put("monitor", "true");
		} catch (Exception e) {
			e.printStackTrace();
			map.put("flag", "false");
			map.put("monitor", "false");
			map.put("msg", e.getMessage());
		}
		Utils.write2PrintWriter(JSON.toJSONString(map));
	}

首先设置所有任务的起始时间，这里往前推迟了10s。是为了防止时间相差太大（也可以设置2s左右。假设tomcat所在机器和集群机器时间同样则不用设置）；接着设置任务的总个数；最后启动多线程执行MR任务。

在任务监控界面，启动一个定时器，会定时向后台请求任务的监控信息，当任务所有完毕则会关闭该定时器。

<script type="text/javascript">
		// 本身主动定时刷新 1s
	 	var monitor_cf_interval= setInterval("monitor_one_refresh()",3000);
	</script>

function monitor_one_refresh(){
	$.ajax({ // ajax提交
		url : 'cloud/cloud_monitorone.action',
		dataType : "json",
		success : function(data) {
			if (data.finished == 'error') {// 获取信息错误 。返回数据设置为0。不然正常返回
				clearInterval(monitor_cf_interval);
				setJobInfoValues(data);
				console.info("monitor,finished:"+data.finished);
				$.messager.show({
					title : '提示',
					msg : '任务执行失败。'
				});
			} else if(data.finished == 'true'){
				// 所有任务执行成功则中止timer
				console.info('monitor,data.finished='+data.finished);
				setJobInfoValues(data);
				clearInterval(monitor_cf_interval);
				$.messager.show({
					title : '提示',
					msg : '所有任务成功执行完毕！
'
				});
				
			}else{
				// 设置提示，并更改页面数据,多行显示job任务信息
				setJobInfoValues(data);
			}
		}
	});
	
}

后台获取任务的监控信息。使用如下的方式：

1）使用JobClient.getAllJobs()获取所有任务的监控信息。

2）使用前面设置的所有任务的启动时间来过滤每个任务；

3）对过滤后的任务依照启动时间进行排序并返回；

4）依据返回任务信息的个数和设置的应该的个数来推断是否中止监控；

/**
	 * 单个任务监控
	 * @throws IOException
	 */
	public void monitorone() throws IOException{
    	Map<String ,Object> jsonMap = new HashMap<String,Object>();
    	List<CurrentJobInfo> currJobList =null;
    	try{
    		currJobList= HUtils.getJobs();
//    		jsonMap.put("rows", currJobList);// 放入数据
    		jsonMap.put("jobnums", HUtils.JOBNUM);
    		// 任务完毕的标识是获取的任务个数必须等于jobNum，同一时候最后一个job完毕
    		// true 所有任务完毕
    		// false 任务正在执行
    		// error 某一个任务执行失败。则再也不监控
    		
    		if(currJobList.size()>=HUtils.JOBNUM){// 假设返回的list有JOBNUM个。那么才可能完毕任务
    			if("success".equals(HUtils.hasFinished(currJobList.get(currJobList.size()-1)))){
    				jsonMap.put("finished", "true");
    				// 执行完毕。初始化时间点
    				HUtils.setJobStartTime(System.currentTimeMillis());
    			}else if("running".equals(HUtils.hasFinished(currJobList.get(currJobList.size()-1)))){
    				jsonMap.put("finished", "false");
    			}else{// fail 或者kill则设置为error
    				jsonMap.put("finished", "error");
    				HUtils.setJobStartTime(System.currentTimeMillis());
    			}
    		}else if(currJobList.size()>0){
    			if("fail".equals(HUtils.hasFinished(currJobList.get(currJobList.size()-1)))||
    					"kill".equals(HUtils.hasFinished(currJobList.get(currJobList.size()-1)))){
    				jsonMap.put("finished", "error");
    				HUtils.setJobStartTime(System.currentTimeMillis());
    			}else{
    				jsonMap.put("finished", "false");
    			}
        	}	
    		if(currJobList.size()==0){
    			jsonMap.put("finished", "false");
//    			return ;
    		}else{
    			if(jsonMap.get("finished").equals("error")){
    				CurrentJobInfo cj =currJobList.get(currJobList.size()-1);
    				cj.setRunState("Error!");
    				jsonMap.put("rows", cj);
    			}else{
    				jsonMap.put("rows", currJobList.get(currJobList.size()-1));
    			}
    		}
    		jsonMap.put("currjob", currJobList.size());
    	}catch(Exception e){
    		e.printStackTrace();
    		jsonMap.put("finished", "error");
    		HUtils.setJobStartTime(System.currentTimeMillis());
    	}
    	System.out.println(new java.util.Date()+":"+JSON.toJSONString(jsonMap));
    	Utils.write2PrintWriter(JSON.toJSONString(jsonMap));// 使用JSON传输数据
    	return ;
    }

获取所有任务，并过滤的代码：

/**
	 * 依据时间来推断，而后得到Job的状态。以此来进行监控 Job的启动时间和使用system.currentTimeMillis得到的时间是一致的。
	 * 不存在时区不一样的问题；
	 * 
	 * @return
	 * @throws IOException
	 */
	public static List<CurrentJobInfo> getJobs() throws IOException {
		JobStatus[] jss = getJobClient().getAllJobs();
		List<CurrentJobInfo> jsList = new ArrayList<CurrentJobInfo>();
		jsList.clear();
		for (JobStatus js : jss) {
			if (js.getStartTime() > jobStartTime) {
				jsList.add(new CurrentJobInfo(getJobClient().getJob(
						js.getJobID()), js.getStartTime(), js.getRunState()));
			}
		}
		Collections.sort(jsList);
		return jsList;
	}

当有多个任务时，使用此监控也是可以的，仅仅用设置HUtils.JOBNUM的值就能够。
3. 文件下载

文件下载便是把过滤后的文件下载到本地，（因为过滤后的文件需要导入到数据库Mysql。因此这里提供下载功能）

文件下载使用FilsSystem.copyToLocalFile()静态方法:

fs.copyToLocalFile(false, file.getPath(), new Path(dst,
							"hdfs_" + (i++) + HUtils.DOWNLOAD_EXTENSION), true);

4.数据入库

数据入库即文件从去重后的本地文件导入到MySql数据库中：

这里使用的是批量插入。同一时候这里不使用xml的解析，而是直接使用字符串的解析，因为在云平台过滤的时候，是去掉了第1，2，最后一行，因此xml文件是不完整的。不能使用xml解析，因此直接使用读取文件，而后进行字符串的解析。

/**
	 * 批量插入xmlPath数据
	 * @param xmlPath
	 * @return
	 */
	public Map<String,Object> insertUserData(String xmlPath){
		Map<String,Object> map = new HashMap<String,Object>();
		try{
			baseDao.executeHql("delete UserData");
//			if(!Utils.changeDat2Xml(xmlPath)){
//				map.put("flag", "false");
//				map.put("msg", "HDFS文件转为xml失败");
//				return map;
//			}
//			List<String[]> strings= Utils.parseXmlFolder2StrArr(xmlPath);
			// ---解析不使用xml解析，直接使用定制解析就能够
			//---
			List<String[]>strings = Utils.parseDatFolder2StrArr(xmlPath);
			List<Object> uds = new ArrayList<Object>();
			for(String[] s:strings){
				uds.add(new UserData(s));
			}
			int ret =baseDao.saveBatch(uds);
			log.info("用户表批量插入了{}条记录!",ret);
		}catch(Exception e){
			e.printStackTrace();
			map.put("flag", "false");
			map.put("msg", e.getMessage());
			return map;
		}
		map.put("flag", "true");
		return map;
	}

public Integer saveBatch(List<Object> lists) {
		Session session = this.getCurrentSession();
//		org.hibernate.Transaction tx = session.beginTransaction();
		int i=0;
		try{
		for ( Object l:lists) {
			i++;
		    session.save(l);
			if( i % 50 == 0 ) { // Same as the JDBC batch size
		        //flush a batch of inserts and release memory:
		        session.flush();
		        session.clear();
		        if(i%1000==0){
		        	System.out.println(new java.util.Date()+"：已经预插入了"+i+"条记录...");
		        }
		    }
		}}catch(Exception e){
			e.printStackTrace();
		}
//		tx.commit();
//		session.close();
		Utils.simpleLog("插入数据数为："+i);
		return i;
	}

5. DB过滤到HDFS

MySQL的用户数据过滤到HDFS，即便用如下的规则进行过滤：

规则 ：reputation>15,upVotes>0,downVotes>0,views>0的用户。

接着。上传这些用户，使用SequenceFile进行写入，因为如下的距离计算便是使用序列化文件做为输入的。因此这里直接写入序列化文件；

private static boolean db2hdfs(List<Object> list, Path path) throws IOException {
		boolean flag =false;
		int recordNum=0;
		SequenceFile.Writer writer = null;
		Configuration conf = getConf();
		try {
			Option optPath = SequenceFile.Writer.file(path);
			Option optKey = SequenceFile.Writer
					.keyClass(IntWritable.class);
			Option optVal = SequenceFile.Writer.valueClass(DoubleArrIntWritable.class);
			writer = SequenceFile.createWriter(conf, optPath, optKey, optVal);
			DoubleArrIntWritable dVal = new DoubleArrIntWritable();
			IntWritable dKey = new IntWritable();
			for (Object user : list) {
				if(!checkUser(user)){
					continue; // 不符合规则 
				}
				dVal.setValue(getDoubleArr(user),-1);
				dKey.set(getIntVal(user));
				writer.append(dKey, dVal);// 用户id,<type,用户的有效向量 >// 后面执行分类的时候需要统一格式，因此这里需要反过来
				recordNum++;
			}
		} catch (IOException e) {
			Utils.simpleLog("db2HDFS失败,+hdfs file:"+path.toString());
			e.printStackTrace();
			flag =false;
			throw e;
		} finally {
			IOUtils.closeStream(writer);
		}
		flag=true;
		Utils.simpleLog("db2HDFS 完毕,hdfs file:"+path.toString()+",records:"+recordNum);
		return flag;
	}

生成文件个数便是HDFS中文件的个数。
6. 距离计算

距离计算即计算每个用户直接的距离。用法即便用两次循环遍历文件，只是这里一共同拥有N*(N-1)/2个输出，因为针对外层用户ID大于内层用户ID的记录，不进行输出。这里使用MR进行。

Mapper的map函数:输出的key-value对是<DoubleWritable,<int,int>>--><距离，<用户i的ID。用户j的ID>>。且用户i的ID<用户j的ID；

public void map(IntWritable key,DoubleArrIntWritable  value,Context cxt)throws InterruptedException,IOException{
		cxt.getCounter(FilterCounter.MAP_COUNTER).increment(1L);
		if(cxt.getCounter(FilterCounter.MAP_COUNTER).getValue()%3000==0){
			log.info("Map处理了{}条记录...",cxt.getCounter(FilterCounter.MAP_COUNTER).getValue());
			log.info("Map生成了{}条记录...",cxt.getCounter(FilterCounter.MAP_OUT_COUNTER).getValue());
		}
		Configuration conf = cxt.getConfiguration();
		SequenceFile.Reader reader = null;
		FileStatus[] fss=input.getFileSystem(conf).listStatus(input);
		for(FileStatus f:fss){
			if(!f.toString().contains("part")){
				continue; // 排除其它文件
			}
			try {
				reader = new SequenceFile.Reader(conf, Reader.file(f.getPath()),
						Reader.bufferSize(4096), Reader.start(0));
				IntWritable dKey = (IntWritable) ReflectionUtils.newInstance(
						reader.getKeyClass(), conf);
				DoubleArrIntWritable dVal = (DoubleArrIntWritable) ReflectionUtils.newInstance(
						reader.getValueClass(), conf);
	
				while (reader.next(dKey, dVal)) {// 循环读取文件
					// 当前IntWritable需要小于给定的dKey
					if(key.get()<dKey.get()){
						cxt.getCounter(FilterCounter.MAP_OUT_COUNTER).increment(1L);
						double dis= HUtils.getDistance(value.getDoubleArr(), dVal.getDoubleArr());
						newKey.set(dis);
						newValue.setValue(key.get(), dKey.get());
						cxt.write(newKey, newValue);
					}

				}
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				IOUtils.closeStream(reader);
			}
		}
	}

Reducer的reduce函数直接输出：

public void reduce(DoubleWritable key,Iterable<IntPairWritable> values,Context cxt)throws InterruptedException,IOException{
		for(IntPairWritable v:values){
			cxt.getCounter(FilterCounter.REDUCE_COUNTER).increment(1);
			cxt.write(key, v);
		}
	}

6. 最佳DC

最佳DC是在”聚类算法“-->”执行聚类“时使用的參数，详细可以參考Clustering by fast search and find of density peaks相关论文。

在寻找最佳DC时是把所有距离依照从大到小进行排序。而后顺序遍历这些距离，取前面的2%左右的数据。这里排序因为在”计算距离“MR任务时，已经利用其Map->reduce的排序性就能够。其距离已经依照距离的大小从小到大排序了，因此仅仅需遍历就能够，这里使用直接遍历序列文件的方式。例如如下：

/**
	 * 依据给定的阈值百分比返回阈值
	 * 
	 * @param percent
	 *            通常为1~2%
	 * @return
	 */
	public static double findInitDC(double percent, String path,long iNPUT_RECORDS2) {
		Path input = null;
		if (path == null) {
			input = new Path(HUtils.getHDFSPath(HUtils.FILTER_CALDISTANCE
					+ "/part-r-00000"));
		} else {
			input = new Path(HUtils.getHDFSPath(path + "/part-r-00000"));
		}
		Configuration conf = HUtils.getConf();
		SequenceFile.Reader reader = null;
		long counter = 0;
		long percent_ = (long) (percent * iNPUT_RECORDS2);
		try {
			reader = new SequenceFile.Reader(conf, Reader.file(input),
					Reader.bufferSize(4096), Reader.start(0));
			DoubleWritable dkey = (DoubleWritable) ReflectionUtils.newInstance(
					reader.getKeyClass(), conf);
			Writable dvalue = (Writable) ReflectionUtils.newInstance(
					reader.getValueClass(), conf);
			while (reader.next(dkey, dvalue)) {// 循环读取文件
				counter++;
				if(counter%1000==0){
					Utils.simpleLog("读取了"+counter+"条记录。。
。");
				}
				if (counter >= percent_) {
					HUtils.DELTA_DC = dkey.get();// 赋予最佳DC阈值
					break;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			IOUtils.closeStream(reader);
		}
		return HUtils.DELTA_DC;
	}

这里需要说明一下，通过试验。发现使用距离阈值29.4时，聚类的决策图中的聚类中心向量并不是十分明显，因此在如下使用的阈值是100；

4.3 聚类算法

1. 执行聚类

执行聚类包含三个MR任务：局部密度MR、最小距离MR以及排序MR任务：

1）局部密度MR

局部密度计算使用的输入文件便是前面计算的距离文件，其MR数据流例如如下：

/**
 * Find the local density of every point vector
 * 
 * 输入为 <key,value>--> <distance,<id_i,id_j>>
 *  <距离，<向量i编号，向量j编号>>
 *  
 *  Mapper：
 *  输出向量i编号，1
 *      向量j编号,1
 *  Reducer:
 *  输出
 *     向量i编号，局部密度
 *     有些向量是没有局部密度的。当某个向量距离其它点的距离所有都大于给定阈值dc时就会发生
 * @author fansy
 * @date 2015-7-3
 */

Mapper的逻辑例如如下：

 

/**
 * 输入为<距离d_ij,<向量i编号，向量j编号>>
 * 依据距离dc阈值推断距离d_ij是否小于dc，符合要求则
 * 输出
 * 向量i编号。1
 * 向量j编号，1
 * @author fansy
 * @date 2015-7-3
 */

map函数：

public void map(DoubleWritable key,IntPairWritable value,Context cxt)throws InterruptedException,IOException{
		double distance= key.get();
		
		if(method.equals("gaussian")){
            one.set(Math.pow(Math.E, -(distance/dc)*(distance/dc)));
        }
		
		if(distance<dc){
			vectorId.set(value.getFirst());
			cxt.write(vectorId, one);
			vectorId.set(value.getSecond());
			cxt.write(vectorId, one);
		}
	}

这里的密度有两种计算方式。依据前台传入的參数选择不一样的算法就能够，这里默认使用的cut-off。即局部密度有一个点则局部密度加1。

reducer中的reduce逻辑即把一样的点的局部密度所有加起来就能够：

public void reduce(IntWritable key, Iterable<DoubleWritable> values,Context cxt)
	throws IOException,InterruptedException{
		double sum =0;
		for(DoubleWritable v:values){
			sum+=v.get();
		}
		sumAll.set(sum);// 
		cxt.write(key, sumAll);
		Utils.simpleLog("vectorI:"+key.get()+",density:"+sumAll);
	}

2）最小距离MR

最小距离MR逻辑例如如下：

/**
 * find delta distance of every point
 * 寻找大于自身密度的最小其它向量的距离
 * mapper输入：
 * 输入为<距离d_ij,<向量i编号，向量j编号>>
 * 把LocalDensityJob的输出
 * 		i,density_i
 * 放入一个map中，用于在mapper中进行推断两个局部密度的大小以决定是否输出
 * mapper输出：
 *      i,<density_i,min_distance_j>
 *      IntWritable,DoublePairWritable
 * reducer 输出：
 * 		<density_i*min_distancd_j> <density_i,min_distance_j,i>
 * 		DoubleWritable,  IntDoublePairWritable
 * @author fansy
 * @date 2015-7-3
 */

这里reducer输出为每个点（即每个用户）局部密度和最小距离的乘积，一种方式寻找聚类中心个数的方法就是把这个乘积从大到小排序，并把这些点画折线图，看其斜率变化最大的点，取前面点的个数即为聚类中心个数。

3）排序MR

排序MR即把2）的局部密度和最小距离的乘积进行排序，这里可以利用map-reduce的排序性，本身定义一个Writable，而后让其依照值的大小从大到小排序。

/**
 * 
 */
package com.fz.fastcluster.keytype;

/**
 * 本身定义DoubleWritable
 * 改动其排序方式，
 * 从大到小排列
 * @author fansy
 * @date 2015-7-3
 */

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

/**
* Writable for Double values.
*/
@InterfaceAudience.Public
@InterfaceStability.Stable
public class CustomDoubleWritable implements WritableComparable<CustomDoubleWritable> {

 private double value = 0.0;
 
 public CustomDoubleWritable() {
   
 }
 
 public CustomDoubleWritable(double value) {
   set(value);
 }
 
 @Override
 public void readFields(DataInput in) throws IOException {
   value = in.readDouble();
 }

 @Override
 public void write(DataOutput out) throws IOException {
   out.writeDouble(value);
 }
 
 public void set(double value) { this.value = value; }
 
 public double get() { return value; }

 /**
  * Returns true iff <code>o</code> is a DoubleWritable with the same value.
  */
 @Override
 public boolean equals(Object o) {
   if (!(o instanceof CustomDoubleWritable)) {
     return false;
   }
   CustomDoubleWritable other = (CustomDoubleWritable)o;
   return this.value == other.value;
 }
 
 @Override
 public int hashCode() {
   return (int)Double.doubleToLongBits(value);
 }
 
 @Override
 public int compareTo(CustomDoubleWritable o) {// 改动这里就能够
   return (value < o.value ?
 1 : (value == o.value ? 0 : -1));
 }
 
 @Override
 public String toString() {
   return Double.toString(value);
 }

 /** A Comparator optimized for DoubleWritable. */ 
 public static class Comparator extends WritableComparator {
   public Comparator() {
     super(CustomDoubleWritable.class);
   }

   @Override
   public int compare(byte[] b1, int s1, int l1,
                      byte[] b2, int s2, int l2) {
     double thisValue = readDouble(b1, s1);
     double thatValue = readDouble(b2, s2);
     return (thisValue < thatValue ?
 1 : (thisValue == thatValue ?
 0 : -1));
   }
 }

 static {                                        // register this comparator
   WritableComparator.define(CustomDoubleWritable.class, new Comparator());
 }

}

2. 画决策图

画决策图，直接解析云平台的排序MR的输出，而后取前面的500条记录（前面500条记录包含的局部密度和最小距离的乘积的最大的500个，后面的点更不可能成为聚类中心点，因此这里仅仅取500个，同一时候需要注意，假设前面设置排序MR的reducer个数大于一个，那么其输出为多个文件，则这里是取每个文件的前面500个向量）

依次点击绘图，展现决策图，就能够看到画出的决策图：

聚类中心应该是取右上角位置的点。因此这里选择去点密度大于50，点距离大于50的点。这里有3个。加上没有画出来的局部密度最大的点。一共同拥有4个聚类中心向量。

3. 寻找聚类中心

寻找聚类中心就是依据前面决策图获得的点密度和点距离阈值来过滤排序MR的输出，获得符合要求的用户ID，这些用户ID便是聚类中心向量的ID。接着，依据这些ID在数据库中找到每个用户ID相应的有效向量（reputation。upVotes，downVotes。views）写入HDFS和本地文件。

写入HDFS是为了做为分类的中心点，写入本地是为了后面查看的方便。

代码例如如下：

/**
	 * 依据给定的阈值寻找聚类中心向量，并写入hdfs
	 * 非MR任务，不需要监控。注意返回值
	 */
	public void center2hdfs(){
		// localfile:method
		// 1. 读取SortJob的输出，获取前面k条记录中的大于局部密度和最小距离阈值的id；
		// 2. 依据id，找到每个id相应的记录；
		// 3. 把记录转为double[] 。
		// 4. 把向量写入hdfs
		// 5. 把向量写入本地文件里，方便后面的查看
		Map<String,Object> retMap=new HashMap<String,Object>();
		
		Map<Object,Object> firstK =null;
		List<Integer> ids= null;
		List<UserData> users=null;
		try{
		firstK=HUtils.readSeq(input==null?HUtils.SORTOUTPUT+"/part-r-00000":input,
				100);// 这里默认使用	前100条记录
		ids=HUtils.getCentIds(firstK,numReducerDensity,numReducerDistance);
		// 2
		users = dBService.getTableData("UserData",ids);
		Utils.simpleLog("聚类中心向量有"+users.size()+"个！");
		// 3,4,5
		HUtils.writecenter2hdfs(users,method,output);	
		}catch(Exception e){
			e.printStackTrace();
			retMap.put("flag", "false");
			retMap.put("msg", e.getMessage());
			Utils.write2PrintWriter(JSON.toJSONString(retMap));
			return ;
		}
		retMap.put("flag", "true");
		Utils.write2PrintWriter(JSON.toJSONString(retMap));
		return ;
	}

写入HDFS和本地的聚类中心例如如下：

4. 执行分类

4.1 执行分类的思路为：

1）聚类中心向量已经写入到_center/iter_0/clustered/part-m-00000中；接着，拷贝原始用户向量（即“DB过滤到HDFS”的输出）到_center/iter_0/unclustered/

2）执行第一次分类。使用Mapper就能够，mapper逻辑为读取_center/iter_0/unclustered/里面的所有文件的每一行。针对每一行A，读取_center/iter_0/clustered/里面所有的数据，循环推断这些向量和A的距离，找到和A的距离最小的距离（同一时候这个距离需要知足大于给定的阈值），并记录这个距离相应向量的类型type，那么就可以输出向量A和类型type，那向量A就已经被分类了，分类后的数据写入到_center/iter_1/clustered里面；假设没有找到最小距离（即所有的距离都大于给定的阈值），那么向量A就是没有被分类的，那么把数据写入到_center/iter_1/unclustered里面。

3）在2）中的mapper中需要记录分类数据和未分类数据的记录数。这样在MR任务执行完毕后。就能够依据这两个数值来推断是否需要进行下次循环。假设这两个数值都是零，那么就退出循环。不然进行下一步。

4）在第i次循环（i>=2）时，使用_center/iter_(i-1)/unclustered里面的数据做为输入，针对这个输入的每一行向量A，遍历_center/iter_1/clustered ~ _center/iter_(i-1)/clustered，使用2）中的方式对A进行分类。假设完毕分类。那么就把数据写入到_center/iter_i/clustered，不然写入到_center/iter_i/unclustered里面。

5）依据第i次MR任务记录的Clustered和Unclustered的值来推断是否进行下次循环。不用则退出循环。不然继续循环进入4）；

map函数代码：

public void map(IntWritable key,DoubleArrIntWritable  value,Context cxt){
		double[] inputI= value.getDoubleArr();
		
		// hdfs
		Configuration conf = cxt.getConfiguration();
		FileSystem fs = null;
		Path path = null;
		
		SequenceFile.Reader reader = null;
		try {
			fs = FileSystem.get(conf);
			// read all before center files 
			String parentFolder =null;
			double smallDistance = Double.MAX_VALUE;
			int smallDistanceType=-1;
			double distance;
			
			// if iter_i !=0,then start i with 1,else start with 0
			for(int i=start;i<iter_i;i++){// all files are clustered points
				
				parentFolder=HUtils.CENTERPATH+"/iter_"+i+"/clustered";
				RemoteIterator<LocatedFileStatus> files=fs.listFiles(new Path(parentFolder), false);
				
				while(files.hasNext()){
					path = files.next().getPath();
					if(!path.toString().contains("part")){
						continue; // return 
					}
					reader = new SequenceFile.Reader(conf, Reader.file(path),
							Reader.bufferSize(4096), Reader.start(0));
					IntWritable dkey = (IntWritable) ReflectionUtils.newInstance(
							reader.getKeyClass(), conf);
					DoubleArrIntWritable dvalue = (DoubleArrIntWritable) ReflectionUtils.newInstance(
							reader.getValueClass(), conf);
					while (reader.next(dkey, dvalue)) {// read file literally
						distance = HUtils.getDistance(inputI, dvalue.getDoubleArr());
					
						if(distance>dc){// not count the farest point
							continue;
						}
						// 这里仅仅要找到离的近期的点并且其distance<=dc 就能够。把这个点的type赋值给当前值就能够
						if(distance<smallDistance){
							smallDistance=distance;
							smallDistanceType=dvalue.getIdentifier();
						}
						
					}// while
				}// while
			}// for
					
			vectorI.set(key.get());// 用户id
			typeDoubleArr.setValue(inputI,smallDistanceType);
			
			if(smallDistanceType!=-1){
				log.info("clustered-->vectorI:{},typeDoubleArr:{}",new Object[]{vectorI,typeDoubleArr.toString()});
				cxt.getCounter(ClusterCounter.CLUSTERED).increment(1);
				out.write("clustered", vectorI, typeDoubleArr,"clustered/part");	
			}else{
				log.info("unclustered---->vectorI:{},typeDoubleArr:{}",new Object[]{vectorI,typeDoubleArr.toString()});
				cxt.getCounter(ClusterCounter.UNCLUSTERED).increment(1);
				out.write("unclustered", vectorI, typeDoubleArr,"unclustered/part");
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			IOUtils.closeStream(reader);
		}

	}

4.2 执行分类的阈值设置

每次循环执行分类时，阈值都是变化的，这里採取的方式是：

1. 计算聚类中心向量两两之间的距离。并依照距离排序，从小到大，每次循环取出距离的一半当作阈值，一直取到最后一个距离；

2. 当进行到K*(K-1)/2个距离时。即最后一个距离（K个聚类中心向量）后。下次循环的阈值设置为当前阈值翻倍。即乘以2；并计数，当再循环k次后，此阈值将再也不变化。

3. 这样设置可以下降误判。同一时候控制循环的次数；

public void run() {
		input=input==null?HUtils.FILTER_PREPAREVECTORS:input;
		
		// 删除iter_i(i>0)的所有文件
		try {
			HUtils.clearCenter((output==null?
HUtils.CENTERPATH:output));
		} catch (FileNotFoundException e2) {
			e2.printStackTrace();
		} catch (IOException e2) {
			e2.printStackTrace();
		}
		
		output=output==null?HUtils.CENTERPATHPREFIX:output+"/iter_";
		
		// 加一个操做。把/user/root/preparevectors里面的数据复制到/user/root/_center/iter_0/unclustered里面
		HUtils.copy(input,output+"0/unclustered");
		try {
			Thread.sleep(200);// 暂停200ms 
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		} 
		
		// 求解dc的阈值。这里的dc不用传入进来就能够，即delta的值
		// 阈值问题可以在讨论，这里临时使用传进来的阈值就能够
//		double dc =dcs[0];
		// 读取聚类中心文件
		Map<Object,Object> vectorsMap= HUtils.readSeq(output+"0/clustered/part-m-00000", Integer.parseInt(k));
		double[][] vectors = HUtils.getCenterVector(vectorsMap);
		double[] distances= Utils.getDistances(vectors);
		// 这里不使用传入进来的阈值
		
		int iter_i=0;
		int ret=0;
		double tmpDelta=0;
		int kInt = Integer.parseInt(k);
		try {
			do{
				if(iter_i>=distances.length){
//					delta= String.valueOf(distances[distances.length-1]/2);
					// 这里使用什么方式尚未想好。。。
					
					
					// 使用如下的方式
					tmpDelta=Double.parseDouble(delta);
					while(kInt-->0){// 超过k次后就再也不增大
						tmpDelta*=2;// 每次翻倍
					}
					delta=String.valueOf(tmpDelta);
				}else{
					delta=String.valueOf(distances[iter_i]/2);
				}
				log.info("this is the {} iteration,with dc:{}",new Object[]{iter_i,delta});
				String[] ar={
						HUtils.getHDFSPath(output)+iter_i+"/unclustered",
						HUtils.getHDFSPath(output)+(iter_i+1),//output
						//HUtils.getHDFSPath(HUtils.CENTERPATHPREFIX)+iter_i+"/clustered/part-m-00000",//center file
						k,
						delta,
						String.valueOf((iter_i+1))
				};
				try{
					ret = ToolRunner.run(HUtils.getConf(), new ClusterDataJob(), ar);
					if(ret!=0){
						log.info("ClusterDataJob failed, with iteration {}",new Object[]{iter_i});
						break;
					}	
				}catch(Exception e){
					e.printStackTrace();
				}
				iter_i++;
				HUtils.JOBNUM++;// 每次循环后加1

			}while(shouldRunNextIter());
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} 
		if(ret==0){
			log.info("All cluster Job finished with iteration {}",new Object[]{iter_i});
		}
		
	}

4.3 执行分类的监控思路

执行监控仍是使用以前的代码，但是这里的MR任务个数一開始并不能直接肯定，那就不能控制监控循环结束的时间。因此这里需要进行改动。这里在MR任务循环完毕以后，设置JOBNUM的值来控制监控任务的结束。并且一開始设置JOBNUM为2。这样在一開始的MR执行结束后就会进行下一次监控循环（这里有个假设就是监控不会仅仅有一次），并且在MR任务每次结束后JOBNUM的值需要递增1：

public void runCluster2(){
		Map<String ,Object> map = new HashMap<String,Object>();
		try {
			//提交一个Hadoop MR任务的基本流程
			// 1. 设置提交时间阈值,并设置这组job的个数
			//使用当前时间就能够,当前时间往前10s。以防server和云平台时间相差
			HUtils.setJobStartTime(System.currentTimeMillis()-10000);// 
			// 因为不知道循环多少次完毕，因此这里设置为2，每次循环都递增1
			// 当所有循环完毕的时候，就该值减去2就能够中止监控部分的循环
			HUtils.JOBNUM=2;
			
			// 2. 使用Thread的方式启动一组MR任务
			new Thread(new RunCluster2(input, output,delta, record)).start();
			// 3. 启动成功后。直接返回到监控，同一时候监控定时向后台获取数据，并在前台展现；
			
			map.put("flag", "true");
			map.put("monitor", "true");
		} catch (Exception e) {
			e.printStackTrace();
			map.put("flag", "false");
			map.put("monitor", "false");
			map.put("msg", e.getMessage());
		}
		Utils.write2PrintWriter(JSON.toJSONString(map));
	}

在MR任务循环结束后，又一次设置JOBNUM的值就能够控制监控的循环中止：

/**
	 * 是否应该继续下次循环
	 * 直接使用分类记录数和未分类记录数来推断
	 * @throws IOException 
	 * @throws IllegalArgumentException 
	 */
	private boolean shouldRunNextIter()  {
		
		if(HUtils.UNCLUSTERED==0||HUtils.CLUSTERED==0){
			HUtils.JOBNUM-=2;// 不用监控 则减去2;
			return false;
		}
		return true;
		
	}

执行分类页面：

这里距离阈值是没实用的，后台直接使用上面的算法获得；循环完毕监控界面的终于阈值，例如如下所看到的：

4.4 聚类中心及推荐

1. 组别入库

组别入库。便是把_center/iter_i/clustered里面的数据解析导入数据库中。导入数据库仍是使用上面的批插入操做：

/**
	 * 把分类的数据解析到list里面
	 * @param path
	 * @return
	 */
	private static Collection<? extends UserGroup> resolve(Path path) {
		// TODO Auto-generated method stub
		List<UserGroup> list = new ArrayList<UserGroup>();
		Configuration conf = HUtils.getConf();
		SequenceFile.Reader reader = null;
		int i=0;
		try {
			reader = new SequenceFile.Reader(conf, Reader.file(path),
					Reader.bufferSize(4096), Reader.start(0));
			IntWritable dkey =  (IntWritable) ReflectionUtils
					.newInstance(reader.getKeyClass(), conf);
			DoubleArrIntWritable dvalue =  (DoubleArrIntWritable) ReflectionUtils
					.newInstance(reader.getValueClass(), conf);

			while (reader.next(dkey, dvalue)) {// 循环读取文件
				// 使用这个进行克隆
				list.add(new UserGroup(i++,dkey.get(),dvalue.getIdentifier()));
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			IOUtils.closeStream(reader);
		}
		Utils.simpleLog("读取"+list.size()+"条记录。文件："+path.toString());
		return list;
	}

2. 聚类中心及占比

聚类中心及占比直接使用数据库中的数据进行统计的，即统计1.中的分类数据每个类别的总记录数，而后再进行计算。聚类中心即直接读取以前写入本地的聚类中心向量文件就能够。

3. 用户查询及推荐

用户查询及推荐即便用用户组内的用户来进行推荐。

依据给定的ID来查询该用户的分组，假设有分组，那么就查询出该分组内的用户，展现到前台：

5. 总结

1. 原始数据通过去重、过滤后，仅剩下541条记录。即对541条记录进行聚类，不算大数据处理。

2. 上面的Hadoop实现的聚类算法，可以使用大数据来測试下，看下效果；

3. 聚类算法在计算两两之间的距离时随着文件的增大，其耗时增加很是快O(N^2)；

4. 使用组内的用户来对用户直接进行推荐的方式有待商榷，可以考虑在组内使用其它方式来过滤用户（比方依据地理位置等信息，该信息在原始数据中是有的），再次推荐；

5. 本项目仅供学习參考，重心在技术和处理方式以及实现方式上；

分享，成长，快乐

脚踏实地，专一

转载请注明blog地址：http://blog.csdn.net/fansy1990