Hadoop MapReduce编程 API入门系列之薪水统计（三十一）
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：输出3-5年经验
//value=record：15-30k
//作为Mapper输出，发给 Reduce 端
//第二步
if(record.length >= 3)//因为取得的薪资在第3列，所以要大于3
{
context.write( new Text(record), new Text(record) );
//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);//过滤数据 //15
int h = filterSalary(arr);//过滤数据 //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);//定义输出路径的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));// 文件输入路径
FileOutputFormat.setOutputPath(job, new Path(args));// 文件输出路径
//第五步：指定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);
}
}
页:
[1]