MapReduce应用案例2
本帖最后由 xuanxufeng 于 2015-8-25 21:20 编辑上一篇：
MapReduce应用案例1
http://www.aboutyun.com/thread-14931-1-1.html
3.10 测试例子10：求任何两名员工信息传递所需要经过的中间节点数3.10.1 问题分析该公司所有员工可以形成入下图的树形结构，求两个员工的沟通的中间节点数，可转换在员工树中求两个节点连通所经过的节点数，即从其中一节点到汇合节点经过节点数加上另一节点到汇合节点经过节点数。例如求M到Q所需节点数，可以先找出M到A经过的节点数，然后找出Q到A经过的节点数，两者相加得到M到Q所需节点数。在作业中首先在Mapper阶段所有员工数据，其中经理数据key为0值、value为"员工编号，员工经理编号"，然后在Reduce阶段把所有员工放到员工列表和员工对应经理链表Map中，最后在Reduce的Cleanup中按照上面说所算法对任意两个员工计算出沟通的路径长度并输出。3.10.2 处理流程图3.10.3 编写代码
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class Q10MiddlePersonsCountForComm extends Configured implements Tool {
public static class MapClass extends Mapper<LongWritable, Text, IntWritable, Text> {
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 对员工文件字段进行拆分
String[] kv = value.toString().split(",");
// 输出key为0和value为员工编号+","+员工经理编号
context.write(new IntWritable(0), new Text(kv + "," + ("".equals(kv) ? " " : kv)));
}
}
public static class Reduce extends Reducer<IntWritable, Text, NullWritable, Text> {
// 定义员工列表和员工对应经理Map
List<String> employeeList = new ArrayList<String>();
Map<String, String> employeeToManagerMap = new HashMap<String, String>();
public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
// 在reduce阶段把所有员工放到员工列表和员工对应经理Map中
for (Text value : values) {
employeeList.add(value.toString().split(",").trim());
employeeToManagerMap.put(value.toString().split(",").trim(), value.toString().split(",").trim());
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
int totalEmployee = employeeList.size();
int i, j;
int distance;
System.out.println(employeeList);
System.out.println(employeeToManagerMap);
// 对任意两个员工计算出沟通的路径长度并输出
for (i = 0; i < (totalEmployee - 1); i++) {
for (j = (i + 1); j < totalEmployee; j++) {
distance = calculateDistance(i, j);
String value = employeeList.get(i) + " and " + employeeList.get(j) + " = " + distance;
context.write(NullWritable.get(), new Text(value));
}
}
}
/**
* 该公司可以由所有员工形成树形结构，求两个员工的沟通的中间节点数，可以转换在员工树中两员工之间的距离
* 由于在树中任意两点都会在某上级节点汇合，根据该情况设计了如下算法
*/
private int calculateDistance(int i, int j) {
String employeeA = employeeList.get(i);
String employeeB = employeeList.get(j);
int distance = 0;
// 如果A是B的经理，反之亦然
if (employeeToManagerMap.get(employeeA).equals(employeeB) || employeeToManagerMap.get(employeeB).equals(employeeA)) {
distance = 0;
}
// A和B在同一经理下
else if(employeeToManagerMap.get(employeeA).equals(
employeeToManagerMap.get(employeeB))) {
distance = 0;
} else {
// 定义A和B对应经理链表
List<String> employeeA_ManagerList = new ArrayList<String>();
List<String> employeeB_ManagerList = new ArrayList<String>();
// 获取从A开始经理链表
employeeA_ManagerList.add(employeeA);
String current = employeeA;
while (false == employeeToManagerMap.get(current).isEmpty()) {
current = employeeToManagerMap.get(current);
employeeA_ManagerList.add(current);
}
// 获取从B开始经理链表
employeeB_ManagerList.add(employeeB);
current = employeeB;
while (false == employeeToManagerMap.get(current).isEmpty()) {
current = employeeToManagerMap.get(current);
employeeB_ManagerList.add(current);
}
int ii = 0, jj = 0;
String currentA_manager, currentB_manager;
boolean found = false;
// 遍历A与B开始经理链表，找出汇合点计算
for (ii = 0; ii < employeeA_ManagerList.size(); ii++) {
currentA_manager = employeeA_ManagerList.get(ii);
for (jj = 0; jj < employeeB_ManagerList.size(); jj++) {
currentB_manager = employeeB_ManagerList.get(jj);
if (currentA_manager.equals(currentB_manager)) {
found = true;
break;
}
}
if (found) {
break;
}
}
// 最后获取两只之前的路径
distance = ii + jj - 1;
}
return distance;
}
}
@Override
public int run(String[] args) throws Exception {
// 实例化作业对象，设置作业名称
Job job = new Job(getConf(), "Q10MiddlePersonsCountForComm");
job.setJobName("Q10MiddlePersonsCountForComm");
// 设置Mapper和Reduce类
job.setJarByClass(Q10MiddlePersonsCountForComm.class);
job.setMapperClass(MapClass.class);
job.setReducerClass(Reduce.class);
// 设置Mapper输出格式类
job.setMapOutputKeyClass(IntWritable.class);
job.setMapOutputValueClass(Text.class);
// 设置Reduce输出键和值类型
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(Text.class);
// 第1个参数为员工数据路径和第2个参数为输出路径
String[] otherArgs = new GenericOptionsParser(job.getConfiguration(), args).getRemainingArgs();
FileInputFormat.addInputPath(job, new Path(otherArgs));
FileOutputFormat.setOutputPath(job, new Path(otherArgs));
job.waitForCompletion(true);
return job.isSuccessful() ? 0 : 1;
}
/**
* 主方法，执行入口
* @param args 输入参数
*/
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new Q10MiddlePersonsCountForComm(), args);
System.exit(res);
}
}
3.10.4 编译并打包代码进入/app/hadoop-1.1.2/myclass/class6目录中新建Q10MiddlePersonsCountForComm.java程序代码（代码页可以使用/home/shiyanlou/install-pack/class6/Q10MiddlePersonsCountForComm.java文件）cd /app/hadoop-1.1.2/myclass/class6vi Q10MiddlePersonsCountForComm.java编译代码javac -classpath ../../hadoop-core-1.1.2.jar:../../lib/commons-cli-1.2.jar Q10MiddlePersonsCountForComm.java把编译好的代码打成jar包，如果不打成jar形式运行会提示class无法找到的错误jar cvf ./Q10MiddlePersonsCountForComm.jar ./Q10MiddlePersons*.classmv *.jar ../..rm Q10MiddlePersons*.class3.10.5 运行并查看结果运行Q10MiddlePersonsCountForComm运行的员工数据路径和输出路径两个参数，需要注意的是hdfs的路径参数路径需要全路径，否则运行会报错：l员工数据路径：hdfs://hadoop:9000/class6/input/empl输出路径：hdfs://hadoop:9000/class6/out10运行如下命令：cd /app/hadoop-1.1.2hadoop jar Q10MiddlePersonsCountForComm.jar Q10MiddlePersonsCountForComm hdfs://hadoop:9000/class6/input/emp hdfs://hadoop:9000/class6/out10运行成功后，刷新CentOS HDFS中的输出路径/class6/out10目录hadoop fs -ls /class6/out10hadoop fs -cat /class6/out10/part-r-00000打开part-r-00000文件，可以看到运行结果：7369 and 7499 = 47369 and 7521 = 47369 and 7566 = 17369 and 7654 = 47369 and 7698 = 3......
作者：石山园出处：http://www.cnblogs.com/shishanyuan/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。
基于MapReduce的应用案例
页:
[1]