MapReduce案例学习（3) 求每个部门最早进入公司的员工姓名

设计思路：

map阶段：将部门名称作为key，将员工姓名和入职时间以逗号分隔拼接成字符串，然后整体作为value输出；

reduce阶段：对map传入的value进行处理，按照逗号切分获得员工姓名和入职时间。因为要求最早进入公司的人员，所以定义一个时间变量，并赋值一个最大的时间值，遍历value时，将员工的入职时间和该最大时间比较，并将该时间变量替换为两者间的较小值。

package week06;
import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
//3) 求每个部门最早进入公司的员工姓名
public class Emp_Test3 extends Configured implements Tool {
/**
* 计数器 用于计数各种异常数据
*/
enum Counter {
LINESKIP,
}
/**
* MAP任务
*/
public static class Map extends Mapper<LongWritable, Text, Text, Text> {
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();// 每行文件
// 输入文件首行，不处理
if (line.contains("empno") == true) {
return;
}
Employee emp = Employee.parser(line);
if (emp.isValid()) {
context.write(new Text(emp.getDname()), new Text(emp.getEname()
+ "," + emp.getHiredate()));
} else {
context.getCounter(Counter.LINESKIP).increment(1); // 出错令计数器+1
return;
}
}
}
/**
* REDUCE任务
*/
public static class Reduce extends Reducer<Text, Text, Text, Text> {
@Override
public void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
String epname = "";
String dateString = "2099-12-25";
Date maxDate = null;
Date hireDt = null;
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
try {
maxDate = sdf.parse(dateString);
} catch (ParseException e) {
System.out.println(e.getMessage());
}
for (Text value : values) {
String[] s = value.toString().split(",");
try {
hireDt = sdf.parse(s[1]);
} catch (ParseException e) {
e.printStackTrace();
}
if (hireDt.before(maxDate)) {
maxDate = hireDt;
epname = s[0];
}
}
context.write(key, new Text(epname + "--" + sdf.format(maxDate)));
}
}
public int run(String[] args) throws Exception {
Configuration conf = getConf();
conf.set("mapred.job.tracker", "192.168.1.201:9001");
String[] ioArgs = new String[] { "emp_in", "emp_out_test3" };
String[] otherArgs = new GenericOptionsParser(conf, ioArgs)
.getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: Test < input path > < output path >");
System.exit(2);
}
Job job = new Job(conf, "week06_test_03"); // 任务名
job.setJarByClass(Emp_Test3.class); // 指定Class
FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 输入路径
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 输出路径
job.setMapperClass(Map.class); // 调用上面Map类作为Map任务代码
job.setReducerClass(Reduce.class);// 调用上面Reduce类作为Reduce任务代码
job.setOutputKeyClass(Text.class); // 指定输出的KEY的格式
job.setOutputValueClass(Text.class); // 指定输出的VALUE的格式
job.waitForCompletion(true);
// 输出任务完成情况
System.out.println("任务名称：" + job.getJobName());
System.out.println("任务成功：" + (job.isSuccessful() ? "是" : "否"));
System.out.println("输入行数："
+ job.getCounters()
.findCounter("org.apache.hadoop.mapred.Task$Counter",
"MAP_INPUT_RECORDS").getValue());
System.out.println("输出行数："
+ job.getCounters()
.findCounter("org.apache.hadoop.mapred.Task$Counter",
"MAP_OUTPUT_RECORDS").getValue());
System.out.println("跳过的行："
+ job.getCounters().findCounter(Counter.LINESKIP).getValue());
return job.isSuccessful() ? 0 : 1;
}
/**
* 设置系统说明 设置MapReduce任务
*/
public static void main(String[] args) throws Exception {
// 记录开始时间
DateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date start = new Date();
// 运行任务
int res = ToolRunner.run(new Configuration(), new Emp_Test3(), args);
// 输出任务耗时
Date end = new Date();
float time = (float) ((end.getTime() - start.getTime()) / 60000.0);
System.out.println("任务开始：" + formatter.format(start));
System.out.println("任务结束：" + formatter.format(end));
System.out.println("任务耗时：" + String.valueOf(time) + " 分钟");
System.exit(res);
}
}

最后

以上就是坚定香水为你收集整理的MapReduce案例学习（3) 求每个部门最早进入公司的员工姓名的全部内容，希望文章能够帮你解决MapReduce案例学习（3) 求每个部门最早进入公司的员工姓名所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。