一、任务目标

MR案例：学生排序（单字段排序、多字段排序）

1. 准备数据

创建sortstudent目录，在里面创建student.txt文件
创建/sortstudent/input目录，执行命令：hdfs dfs -mkdir -p /sortstudent/input
将文本文件student.txt，上传到HDFS的/sortstudent/input目录

二、实行任务

1. 创建Maven项目

Maven项目 - SortStudent

2. 添加相关依赖

在pom.xml文件里添加hadoop和junit依赖

<dependencies>                                      <!--hadoop客户端-->                                <dependency>                                    <groupId>org.apache.hadoop</groupId>        <artifactId>hadoop-client</artifactId>      <version>3.3.4</version>                    </dependency>                                   <!--单元测试框架-->                                   <dependency>                                    <groupId>junit</groupId>                    <artifactId>junit</artifactId>              <version>4.13.2</version>                   </dependency>                                   
</dependencies>

3. 创建日志属性文件

在resources目录里创建log4j.properties文件

log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/sortstudent.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

4. 创建学生实体类

在net.kox.mr包里创建Student类

package net.kox.mr;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class Student implements WritableComparable<Student> {private String name;private String gender;private int age;private String phone;private String major;public String getName() {return name;}public void setName(String name) {this.name = name;}public String getGender() {return gender;}public void setGender(String gender) {this.gender = gender;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}public String getPhone() {return phone;}public void setPhone(String phone) {this.phone = phone;}public String getMajor() {return major;}public void setMajor(String major) {this.major = major;}@Overridepublic String toString() {return "Student{" +"name='" + name + '\'' +", gender='" + gender + '\'' +", age=" + age +", phone='" + phone + '\'' +", major='" + major + '\'' +'}';}public int compareTo(Student o) {return o.getAge() - this.getAge(); // 降序}public void write(DataOutput out) throws IOException {out.writeUTF(name);out.writeUTF(gender);out.writeInt(age);out.writeUTF(phone);out.writeUTF(major);}public void readFields(DataInput in) throws IOException {name = in.readUTF();gender = in.readUTF();age = in.readInt();phone = in.readUTF();major = in.readUTF();}
}

5. 创建学生映射器类

在net.kox.mr里创建StudentMapper类

package net.kox.mr;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class StudentMapper extends Mapper<LongWritable, Text, Student, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// 获取行内容String line = value.toString();// 按空格拆分得到字段数组String[] fields = line.split(" ");// 获取学生信息String name = fields[0];String gender = fields[1];int age = Integer.parseInt(fields[2]);String phone = fields[3];String major = fields[4];// 创建学生对象Student student = new Student();// 设置学生对象属性student.setName(name);student.setGender(gender);student.setAge(age);student.setPhone(phone);student.setMajor(major);// 写入键值对<student,null>context.write(student, NullWritable.get());}
}

6. 创建学生归并器类

package net.kox.mr;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class StudentReducer extends Reducer<Student, NullWritable, Text, NullWritable> {@Overrideprotected void reduce(Student key, Iterable<NullWritable> values, Context context)throws IOException, InterruptedException {// 获取学生对象Student student = key;// 拼接学生信息String studentInfo = student.getName() + "\t"+ student.getGender() + "\t"+ student.getAge() + "\t"+ student.getPhone() + "\t"+ student.getMajor();// 写入键值对<studentInfo,null>context.write(new Text(studentInfo), NullWritable.get());}
}

7. 创建学生驱动类

在net.kox.mr包里创建StudentDriver类

package net.kox.mr;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
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;import java.net.URI;public class StudentDriver {public static void main(String[] args) throws Exception {// 创建配置对象Configuration conf = new Configuration();// 设置数据节点主机名属性conf.set("dfs.client.use.datanode.hostname", "true");// 获取作业实例Job job = Job.getInstance(conf);// 设置作业启动类job.setJarByClass(StudentDriver.class);// 设置Mapper类job.setMapperClass(StudentMapper.class);// 设置map任务输出键类型job.setMapOutputKeyClass(Student.class);// 设置map任务输出值类型job.setMapOutputValueClass(NullWritable.class);// 设置Reducer类job.setReducerClass(StudentReducer.class);// 设置reduce任务输出键类型job.setOutputKeyClass(Student.class);// 设置reduce任务输出值类型job.setOutputValueClass(NullWritable.class);// 定义uri字符串String uri = "hdfs://master:9000";// 创建输入目录Path inputPath = new Path(uri + "/sortstudent/input");// 创建输出目录Path outputPath = new Path(uri + "/sortstudent/output");// 获取文件系统FileSystem fs =  FileSystem.get(new URI(uri), conf);// 删除输出目录（第二个参数设置是否递归）fs.delete(outputPath, true);// 给作业添加输入目录（允许多个）FileInputFormat.addInputPath(job, inputPath);// 给作业设置输出目录（只能一个）FileOutputFormat.setOutputPath(job, outputPath);// 等待作业完成job.waitForCompletion(true);// 输出统计结果System.out.println("======统计结果======");FileStatus[] fileStatuses = fs.listStatus(outputPath);for (int i = 1; i < fileStatuses.length; i++) {// 输出结果文件路径System.out.println(fileStatuses[i].getPath());// 获取文件系统数据字节输入流FSDataInputStream in = fs.open(fileStatuses[i].getPath());// 将结果文件显示在控制台IOUtils.copyBytes(in, System.out, 4096, false);}}
}