MapReduce基础案例06-计算平均成绩
在本教程中，我们将通过编写MapReduce来统计各科目的平均成绩。
我们需要用到三个成绩数据文件。
Math内容为：语文成绩 china.txt。内容如下：
zhangsan 78 lisi 89 wangwu 96 zhaoliu 67
英语成绩 english.txt。内容如下：
zhangsan 80 lisi 82 wangwu 84 zhaoliu 86
数学成绩 math.txt。内容如下：
zhangsan 88 lisi 99 wangwu 66 zhaoliu 77
一、创建Java Maven项目
Maven依赖：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>HadoopDemo</groupId>
<artifactId>com.xueai8</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<!--hadoop依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.1</version>
</dependency>
<!--hdfs文件系统依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>3.3.1</version>
</dependency>
<!--MapReduce相关的依赖-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>3.3.1</version>
</dependency>
<!--junit依赖-->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<!--编译器插件用于编译拓扑-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source> <!-- 源代码使用的JDK版本 -->
<target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
<encoding>UTF-8</encoding><!-- 字符集编码 -->
</configuration>
</plugin>
</plugins>
</build>
</project>
AvgMapper.java：
package com.xueai8.avg;
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class AvgMapper extends Mapper<LongWritable, Text, Text, FloatWritable> {
// 定义可重用的key和value对象
private final static Text course = new Text();
private final static FloatWritable score = new FloatWritable(0);
@Override
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 从hadoop Text类型转换为 java String类型
String line = value.toString();
// 对读取的每一行文本进行分词
StringTokenizer tokenizerLine = new StringTokenizer(line);
String strName = tokenizerLine.nextToken(); // 科目
String strScore = tokenizerLine.nextToken(); // 成绩
course.set(strName);
score.set(Float.parseFloat(strScore));
context.write(course, score);
}
}
AvgReducer.java
package com.xueai8.avg;
import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class AvgReducer extends Reducer<Text, FloatWritable, Text, FloatWritable> {
// 定义可重用的key和value对象
private final static Text course = new Text();
private final static FloatWritable avgScore = new FloatWritable(0);
@Override
public void reduce(Text key, Iterable<FloatWritable> values, Context context)
throws IOException, InterruptedException {
float sum = 0; // 记录某一科目总成绩
float count = 0; // 记录某一科目学生总数
for (FloatWritable value : values) {
sum += value.get(); // 累加总成绩
count++; // 统计成绩数量
}
float average = sum / count; // 计算平均成绩
course.set(key);
// avgScore.set(average);
avgScore.set(Math.round(average*100)/100.00f); // 保留小数点后两位
context.write(course, avgScore);// 写出
}
}
AvgDriver.java：
package com.xueai8.avg;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class AvgDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("用法: AvgDriver <in> <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "Score Average");
job.setJarByClass(AvgDriver.class);
// 设置Mapper
job.setMapperClass(AvgMapper.class);
// Combiner
job.setCombinerClass(AvgReducer.class);
// Reducer
job.setReducerClass(AvgReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FloatWritable.class);
// 输入输出格式类
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
// 输入输出路径
FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
// 提交作业
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
二、配置log4j
在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：
log4j.rootLogger = info,stdout
### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n
三、项目打包
打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：
如果一切正常，会提示打jar包成功。如下图所示：
这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：
四、项目部署
请按以下步骤执行。
1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：
$ start-dfs.sh
$ start-yarn.sh
查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：
$ jps
这时应该能看到有如下5个进程正在运行，说明集群运行正常：
5542 NodeManager
5191 SecondaryNameNode
4857 NameNode
5418 ResourceManager
4975 DataNode
2、将数据文件sample.txt上传到HDFS的/data/mr/目录下。
$ hdfs dfs -mkdir -p /data/mr $ hdfs dfs -put china.txt /data/mr/ $ hdfs dfs -put english.txt /data/mr/ $ hdfs dfs -put math.txt /data/mr/ $ hdfs dfs -ls /data/mr/
3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）
在终端窗口中，执行如下的作业提交命令：
$ hadoop jar com.xueai8-1.0-SNAPSHOT.jar com.xueai8.avg.AvgDriver /data/mr /data/mr-output
4、查看输出结果。
在终端窗口中，执行如下的HDFS命令，查看输出结果：
$ hdfs dfs -ls /data/mr-output $ hdfs dfs -cat /data/mr-output/part-r-00000
可以得到类似下面这样的输出结果：
lisi 90.0 wangwu 82.0 zhangsan 82.0 zhaoliu 76.67
