Hadoop经典案例Spark实现（四）——平均成绩

87 阅读 0 评论 58 点赞

我是靠谱客的博主秀丽嚓茶，这篇文章主要介绍Hadoop经典案例Spark实现（四）——平均成绩，现在分享给大家，希望可以做个参考。

Hadoop经典案例Spark实现（四）——平均成绩

一、案例分析：
1、需求分析
对输入文件中数据进行就算学生平均成绩。输入文件中的每行内容均为一个学生的姓名和他相应的成绩，如果有多门学科，则每门学科为一个文件。
要求在输出中每行有两个间隔的数据，其中，第一个代表学生的姓名，第二个代表其平均成绩。

2、原始数据
1）math：

复制代码

1
2
3
4
5
6
7
8
张三
88
李四
99
王五
66
赵六
77

2）china：

复制代码

1
2
3
4
5
6
7
8
张三
78
李四
89
王五
96
赵六
67

3）english：

复制代码

1
2
3
4
5
6
7
8
张三
80
李四
82
王五
84
赵六
86

样本输出：

复制代码

3、设计思考
Map处理的是一个纯文本文件，文件中存放的数据时每一行表示一个学生的姓名和他相应一科成绩。Mapper处理的数据是由InputFormat分解过的数据集，
其中 InputFormat的作用是将数据集切割成小数据集InputSplit，每一个InputSplit将由一个Mapper负责处理。此外，InputFormat中还提供了一个RecordReader的实现，
并将一个InputSplit解析成<key,value>对提供给了map函数。InputFormat的默认值是TextInputFormat，它针对文本文件，按行将文本切割成InputSlit，
并用 LineRecordReader将InputSplit解析成<key,value>对，key是行在文本中的位置，value是文件中的一行。

Map的结果会通过partion分发到Reducer，Reducer做完Reduce操作后，将通过以格式OutputFormat输出。

Mapper最终处理的结果对<key,value>，会送到Reducer中进行合并，合并的时候，有相同key的键/值对则送到同一个 Reducer上。
Reducer是所有用户定制Reducer类地基础，它的输入是key和这个key对应的所有value的一个迭代器，同时还有 Reducer的上下文。

Reduce的结果由Reducer.Context的write方法输出到文件中。

二、MapReduce实现

Map代码

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import java.io.IOException;
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, IntWritable> {
@Override
protected void map(LongWritable key, Text value,Context context)
throws IOException, InterruptedException {
String line = value.toString();
if(line.trim().length()>0){
String[] arr = line.split("t");
if(arr.length==2){
context.write(new Text(arr[0]), new IntWritable(Integer.valueOf(arr[1])));
}
}
}
}

Reduce代码

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import java.io.IOException;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class AvgReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
@Override
protected void reduce(Text key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
int sum = 0;
for(IntWritable val:values){
sum += val.get();
}
context.write(key, new DoubleWritable(sum/3.0));
}
}

Job提交

复制代码

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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 JobMain {
/**
* @param args
*/
public static void main(String[] args)throws Exception {
Configuration configuration = new Configuration();
Job job = new Job(configuration,"avg-job");
job.setJarByClass(JobMain.class);
job.setMapperClass(AvgMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setReducerClass(AvgReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileSystem fs = FileSystem.get(configuration);
Path outputDir = new Path(args[1]);
if(fs.exists(outputDir)){
fs.delete(outputDir, true);
}
FileOutputFormat.setOutputPath(job, outputDir);
System.exit(job.waitForCompletion(true)?0:1);
}
}

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
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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 JobMain {
/**
* @param args
*/
public static void main(String[] args)throws Exception {
Configuration configuration = new Configuration();
Job job = new Job(configuration,"avg-job");
job.setJarByClass(JobMain.class);
job.setMapperClass(AvgMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
job.setReducerClass(AvgReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileSystem fs = FileSystem.get(configuration);
Path outputDir = new Path(args[1]);
if(fs.exists(outputDir)){
fs.delete(outputDir, true);
}
FileOutputFormat.setOutputPath(job, outputDir);
System.exit(job.waitForCompletion(true)?0:1);
}
}

三、Spark实现-Scala版本

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
val fourth = sc.textFile("/tmp/spark/fourth",3)
val res = fourth.filter(_.trim().length>0).map(line=>(line.split("t")(0).trim(),line.split("t")(1).trim().toInt)).groupByKey().map(x => {
var num = 0.0
var sum = 0
for(i <- x._2){
sum = sum + i
num = num +1
}
val avg = sum/num
val format = f"$avg%1.2f".toDouble
(x._1,format)
}).collect.foreach(x => println(x._1+"t"+x._2))

思路：先groupBy分组，再map处理成绩的集合，并做了格式化显示

结果是一样