MapReduce程式設計筆記(3)-計算部門工資
阿新 • • 發佈:2021-08-01
一、分析資料處理的過程
二、程式程式碼
2.1 main程式
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class SalaryTotalMain { public static void main(String[] args) throws Exception { //1、建立任務Job,並且指定任務的入口 Job job = Job.getInstance(new Configuration()); job.setJarByClass(SalaryTotalMain.class); //2、指定任務的Map,Map的輸出型別 job.setMapperClass(SalaryTotalMapper.class); job.setMapOutputKeyClass(IntWritable.class); //k2 job.setMapOutputValueClass(IntWritable.class); //v2 //3、指定任務的Reduce,Reduce的輸出型別 job.setReducerClass(SalaryTotalReducer.class); job.setOutputKeyClass(IntWritable.class); //k4 job.setOutputValueClass(IntWritable.class); //v4 //4、指定任務的輸入路徑和輸出路徑 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); //5、執行任務 job.waitForCompletion(true); } }
2.2 Map程式
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SalaryTotalMapper extends Mapper<LongWritable, Text,IntWritable, IntWritable> { @Override protected void map(LongWritable key1, Text value1, Context context) throws IOException, InterruptedException { /* context代表Map的上下文 上文:DHFS的輸入 下文:Reduce */ String data =value1.toString(); String [] words = data.split(","); for ( String w:words){ context.write(new IntWritable(Integer.parseInt(words[7])),new IntWritable(Integer.parseInt(words[5]))); } } }
2.1 Reduce程式
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SalaryTotalReducer extends Reducer<IntWritable, IntWritable,IntWritable,IntWritable> { @Override protected void reduce(IntWritable k3, Iterable<IntWritable> v3, Context context) throws IOException, InterruptedException { int total = 0; for (IntWritable v:v3){ total += v.get(); } context.write(k3,new IntWritable(total)); } }