Hadoop MapReduce编程 API入门系列之薪水统计（三十一）

fumingxia

package zhouls.bigdata.myMapReduce.SalaryCount;　　

　　
import java.io.IOException;
　　

　　
import java.util.regex.Pattern;
　　
import org.apache.hadoop.conf.Configuration;
　　
import org.apache.hadoop.conf.Configured;
　　
import org.apache.hadoop.fs.FileSystem;
　　
import org.apache.hadoop.fs.Path;
　　
import org.apache.hadoop.io.Text;
　　
import org.apache.hadoop.io.LongWritable;
　　
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.util.Tool;
　　
import org.apache.hadoop.util.ToolRunner;
　　

　　

　　

/**　　
* 基于样本数据做Hadoop工程师薪资统计：计算各工作年限段的薪水范围
　　

*/
　　
public>　　
{

　　
public static>　　
{
　　
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
　　
{
　　
// 美团 3-5年经验 15-30k 北京 【够牛就来】hadoop高级工程...
　　
//北信源 3-5年经验 15-20k 北京 Java高级工程师（有Hadoo...
　　
// 蘑菇街 3-5年经验 10-24k 杭州 hadoop开发工程师
　　

　　
//第一步，将输入的纯文本文件的数据转化成String
　　
String line = value.toString();//读取每行数据
　　

　　
String[] record = line.split( "\\s+");//使用空格正则解析数据
　　
//key=record[1]：输出3-5年经验
　　
//value=record[2]：15-30k
　　
//作为Mapper输出，发给 Reduce 端
　　

　　
//第二步
　　
if(record.length >= 3)//因为取得的薪资在第3列，所以要大于3
　　
{
　　
context.write( new Text(record[1]), new Text(record[2]) );
　　
//Map输出，record数组的第2列，第3列
　　
}
　　
}
　　
}

　　
public static>　　
{
　　
public void reduce(Text Key, Iterable<Text> Values, Context context) throws IOException, InterruptedException
　　
{
　　

　　
int low = 0;//记录最低工资
　　
int high = 0;//记录最高工资
　　
int count = 1;
　　
//针对同一个工作年限（key），循环薪资集合（values），并拆分value值，统计出最低工资low和最高工资high
　　
for (Text value : Values)
　　
{
　　
String[] arr = value.toString().split("-");//其中的一行而已，15 30K
　　
int l = filterSalary(arr[0]);//过滤数据 //15
　　
int h = filterSalary(arr[1]);//过滤数据 //30
　　
if(count==1 || l< low)
　　
{
　　
low = l;
　　
}
　　
if(count==1 || h>high)
　　
{
　　
high = h;
　　
}
　　
count++;
　　
}
　　
context.write(Key, new Text(low + "-" +high + "k"));//即10-30K
　　
}
　　
}
　　
//正则表达式提取工资值，因为15 30k,后面有k,不干净
　　
public static int filterSalary(String salary)//过滤数据
　　
{
　　
String sal = Pattern.compile("[^0-9]").matcher(salary).replaceAll("");
　　
return Integer.parseInt(sal);
　　
}
　　

　　

　　
public int run(String[] args) throws Exception
　　
{
　　
//第一步：读取配置文件
　　
Configuration conf = new Configuration();//读取配置文件
　　

　　
//第二步:输出路径存在就先删除
　　
Path out = new Path(args[1]);//定义输出路径的Path对象，mypath
　　
FileSystem hdfs = out.getFileSystem(conf);//通过路径下的getFileSystem来获得文件系统
　　
if (hdfs.isDirectory(out))
　　
{//删除已经存在的输出目录
　　
hdfs.delete(out, true);
　　
}
　　
//第三步：构建job对象
　　
Job job = new Job(conf, "SalaryCount" );//新建一个任务
　　
job.setJarByClass(SalaryCount.class);//设置 主类
　　
//通过job对象来设置主类SalaryCount.class
　　

　　
//第四步：指定数据的输入路径和输出路径
　　
FileInputFormat.addInputPath(job, new Path(args[0]));// 文件输入路径
　　
FileOutputFormat.setOutputPath(job, new Path(args[1]));// 文件输出路径
　　

　　
//第五步：指定Mapper和Reducer
　　
job.setMapperClass(SalaryMapper.class);// Mapper
　　
job.setReducerClass(SalaryReducer.class);// Reducer
　　

　　
//第六步：设置map函数和reducer函数的输出类型
　　
job.setOutputKeyClass(Text.class);//输出结果key类型
　　
job.setOutputValueClass(Text.class);//输出结果的value类型
　　

　　
//第七步：提交作业
　　
job.waitForCompletion(true);//等待完成退出作业
　　

　　
return 0;
　　
}
　　

　　

　　
/**
　　
* @param args 输入文件、输出路径，可在Eclipse中Run Configurations中配Arguments，如：
　　
* hdfs://HadoopMaster:9000/salary.txt
　　
* hdfs://HadoopMaster:9000/out/salary
　　
*/
　　
public static void main(String[] args) throws Exception
　　
{
　　
//第一步
　　
String[] args0 =
　　
{
　　
// "hdfs://HadoopMaster:9000/salary/",
　　
// "hdfs://HadoopMaster:9000/out/salary/"
　　
"./data/salary/salary.txt",
　　
"./out/salary"
　　
};
　　
//第二步
　　
int ec = ToolRunner.run(new Configuration(), new SalaryCount(), args0);
　　
//第一个参数是读取配置文件，第二个参数是主类Temperature，第三个参数是输入路径和输出路径的属组
　　
System.exit(ec);
　　

　　
}
　　
}