分组汇总

Mapper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package cn.studybigdata.hadoop.mapred.bdeptsalary;

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 DeptMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
    //7369,SMITH,CLERK,7902,1980/12/17,800,,20
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String lineContent = value.toString();
        String[] empArray = lineContent.split(",");

        int dept = Integer.parseInt(empArray[7]);
        int salary = Integer.parseInt(empArray[5]);

        // key: 部门
        // value: 工资
        context.write(new IntWritable(dept), new IntWritable(salary));
    }
}

Reducer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package cn.studybigdata.hadoop.mapred.bdeptsalary;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class DeptReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
    @Override
    protected void reduce(IntWritable dept, Iterable<IntWritable> salaries, Context context) throws IOException, InterruptedException {

        int total = 0;

        for (IntWritable salary : salaries) {
            int i = salary.get();
            total = total + i;
        }    
    
        context.write(dept, new IntWritable(total));

    }
}

Main

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
package cn.studybigdata.hadoop.mapred.bdeptsalary;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class DeptMain {

    //hadoop jar xxx.jar arg0 arg1
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        Configuration configuration = new Configuration();
        //create a mapreduce job
        Job job = Job.getInstance(configuration);


        job.setJarByClass(DeptMain.class);

        job.setMapperClass(DeptMapper.class);

        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);


        job.setReducerClass(DeptReduce.class);
        //<word,3>
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);

        //input path
        //output path
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);

    }
}

CLI arguments

1
D:\IdeaProjects\studybigdata\data\emp\emp.csv D:\IdeaProjects\studybigdata\hadoop\target\out