【李老师云计算】作业二:参照Eclipse Mapreduc访问Hadoop文档,求解n个数的最大值
持续更新,直到结束……内容可能来自博主自己手搓、吸取同学的经验、网络上内容的整合等等,仅供参考,更多内容可以查看大三下速通指南专栏。
前言
代码在最后,和实验一一样的操作换个代码就行了。
内容可能来自博主自己手搓、吸取同学的经验、网络上内容的整合等等,仅供参考,更多内容可以查看大三下速通指南专栏。
参照Eclipse Mapreduc访问Hadoop文档,求解n个数的最大值
注意按理来说这个作业要完成实验一才能做,但是不知为何实验一的提交时间是晚于这个作业的(晚一个周),不知道是什么用意,可能是为了让我们沉淀一下,迎接审判。
完整的代码在最下面,如果运行Hadoop的项目不把从机打开,可能会出现出现目录但没有结果文件的情况。虽然是这么写的,但不一定要按照这个来,那些描述都是跑出来的可以看一下但不要较真……
题目描述:
-
MapReduce介绍
1.1 MapReduce编程模型(评分点1)
1.2 MapReduce处理过程(评分点2)
-
MapReduce求最大值处理过程
2.1 分割过程(评分点3)
2.2 Map排序与Combine过程(评分点4)
2.3 Reduce排序与输出结果(评分点5)
-
基于Eclipse的MapReduce项目求解最大值
3.1 Eclipse访问HDFS(评分点6)
3.2 作业配置实现(评分点7)编写main函数,辅以充分的注释
3.3 Map过程的实现(评分点8)Mapper类的局部变量、编写map函数,辅以充分的注释
3.4 Reduce过程的实现(评分点9)Reducer类的局部变量、编写reduce函数,辅以充分的注释
-
测试(评分点10)给出测试数据,运行程序,得出计算结果。
参考答案:
1.1 MapReduce编程模型
MapReduce编程模型是一种分布式计算框架,其核心思想是将大规模数据集划分为许多小数据块,然后将这些小数据块分别交给多个计算节点进行处理,最终将结果进行合并。MapReduce编程模型包含两个阶段:Map阶段和Reduce阶段。在Map阶段,每个计算节点都会对自己所负责的数据块进行处理,将其映射为一系列键值对。在Reduce阶段,这些键值对会被按照键进行合并,并进行聚合操作。
1.2 MapReduce处理过程
MapReduce处理过程包括以下几个步骤:
- 输入分片:将输入数据分割为若干个数据块。
- Map处理:对每个数据块进行Map处理,并将结果输出为若干个键值对。
- Shuffle过程:对Map输出的键值对进行合并、排序、分组等操作,将同一键的值合并在一起,生成若干个键值对集合。
- Reduce处理:对每个键值对集合进行Reduce处理,生成若干个输出键值对。
- 输出:将Reduce输出的结果保存到输出文件中。
- MapReduce求最大值处理过程
2.1 分割过程
在这个问题中,我们需要将输入数据分割为若干个数据块,每个数据块包含若干个数值。由于我们需要求的是最大值,因此我们可以将数据块分割为大小相等的若干个子数据块,每个子数据块的大小为m。例如,如果输入数据为[1, 3, 5, 2, 4, 6, 7, 9, 8, 10],并且m=3,则我们可以将数据块分割为三个子数据块:[1, 3, 5], [2, 4, 6], [7, 9, 8, 10]。
2.2 Map排序与Combine过程
在Map阶段,每个计算节点都会对自己所负责的数据块进行处理,将其映射为一系列键值对。由于我们需要求最大值,因此我们可以将每个数据块中的最大值作为键,将输入数据块的编号作为值。例如,对于子数据块[1, 3, 5],其最大值为5,编号为0,因此我们可以将键值对(5, 0)输出。
在Map阶段的输出结果需要进行合并操作,以减少Reduce阶段的数据量。在本问题中,我们可以将Map输出的键值对按照键进行排序,并将同一键的值进行合并,生成若干个键值对集合。由于我们的键是数值,因此我们需要对键进行升序排序。
在对键值对进行排序的过程中,可以通过自定义比较器来指定排序方式。对于本问题中的键值对,我们可以通过编写一个比较器类来指定按照键的升序排序。下面是一个示例代码:
public class MaxValueComparator extends WritableComparator {
protected MaxValueComparator() {
super(IntWritable.class, true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
IntWritable int1 = (IntWritable) a;
IntWritable int2 = (IntWritable) b;
return int1.compareTo(int2);
}
}
在Map阶段的输出结果进行合并操作时,可以通过Combine过程来进一步减少数据量。Combine过程与Reduce过程相似,但是其运行在Map节点上,并且可以在Map节点上进行合并操作。在本问题中,我们可以使用Reduce过程作为Combine过程,将同一键的值进行合并,并输出合并后的键值对集合。
2.3 Reduce排序与输出结果
在Reduce阶段,对于每个键值对集合,我们可以对其进行Reduce操作,求出最大值。例如,对于键值对集合(5, [0, 2]),我们可以求出其对应的最大值10。对于本问题中的Reduce操作,我们可以编写一个Reducer类来实现。下面是一个示例代码:
public class MaxValueReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
private IntWritable result = new IntWritable();
@Override
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int max = Integer.MIN_VALUE;
for (IntWritable value : values) {
max = Math.max(max, value.get());
}
result.set(max);
context.write(key, result);
}
}
在Reduce阶段的输出结果需要进行排序操作,以便将最大值排在最前面。在本问题中,我们可以使用与Map阶段类似的方式对输出结果进行排序,即使用一个比较器类来指定按照键的升序排序。下面是一个示例代码:
job.setSortComparatorClass(MaxValueComparator.class);
- 基于Eclipse的MapReduce项目求解最大值
3.1 Eclipse访问HDFS
在基于Eclipse的MapReduce项目中,我们需要使用HDFS作为输入和输出的数据源。因此,在编写代码之前,我们需要确保Hadoop和HDFS已经安装并运行在本地环境中。在Eclipse中,可以通过添加Hadoop库来访问HDFS。下面是一个示例代码:
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");
FileSystem fs = FileSystem.get(conf);
Path inputPath = new Path("/input");
Path outputPath = new Path("/output");
if (fs.exists(outputPath)) {
fs.delete(outputPath, true);
}
3.2 作业配置实现
在配置MapReduce作业时,我们需要指定作业的输入、输出、Mapper类、Reducer类等信息。
Job job = Job.getInstance(conf, "max value");
job.setJarByClass(MaxValue.class);
job.setMapperClass(MaxValueMapper.class);
job.setCombinerClass(MaxValueReducer.class);
job.setReducerClass(MaxValueReducer.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
在配置作业时,我们需要通过Job类的getInstance方法来获取一个作业实例,并指定作业的名称和运行配置。在本问题中,我们可以指定作业名称为"max value"。通过setJarByClass方法来指定运行作业的类,即MaxValue类。通过setMapperClass、setCombinerClass和setReducerClass方法来指定Mapper、Combine和Reducer类。通过setOutputKeyClass和setOutputValueClass方法来指定输出键和值的类型。最后,通过FileInputFormat和FileOutputFormat类的addInputPath和setOutputPath方法来指定作业的输入和输出路径。
public class MaxValue {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://localhost:9000");
FileSystem fs = FileSystem.get(conf);
Path inputPath = new Path("/input");
Path outputPath = new Path("/output");
if (fs.exists(outputPath)) {
fs.delete(outputPath, true);
}
Job job = Job.getInstance(conf, "max value");
job.setJarByClass(MaxValue.class);
job.setMapperClass(MaxValueMapper.class);
job.setCombinerClass(MaxValueReducer.class);
job.setReducerClass(MaxValueReducer.class);
job.setOutputKeyClass(IntWritable.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, inputPath);
FileOutputFormat.setOutputPath(job, outputPath);
boolean success = job.waitForCompletion(true);
if (success) {
System.out.println("Job completed successfully.");
}
}
}
在main函数中,我们首先创建一个Configuration对象,并设置默认文件系统为本地HDFS。然后,我们获取一个FileSystem对象,并指定输入和输出路径。在作业配置之后,我们通过调用waitForCompletion方法来等待作业运行完毕。最后,我们输出作业运行结果。如果作业运行成功,输出"Job completed successfully."。
3.3 Map过程的实现
public class MaxValueMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> {
private final IntWritable one = new IntWritable(1);
private IntWritable number = new IntWritable();
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] numbers = value.toString().split(",");
for (String s : numbers) {
number.set(Integer.parseInt(s));
context.write(one, number);
}
}
}
在Mapper类中,我们首先声明两个局部变量:一个IntWritable类型的one变量,用于作为键;一个IntWritable类型的number变量,用于作为值。在map函数中,我们首先将输入的一行文本转换为一个字符串数组,然后遍历该数组。对于数组中的每个元素,我们将其转换为一个整数,并将其赋值给number变量。接下来,我们将one作为键,number作为值,通过调用Context对象的write方法写入上下文。这样,Map函数就将每个输入数值作为值输出,而将固定的键1与每个数值组合。
在Reducer类中,我们同样首先声明一个局部变量,用于保存输入值的最大值。在reduce函数中,对于每个键值对,我们将值转换为一个整数,并与当前最大值进行比较。如果值大于当前最大值,则将该值赋值给最大值变量。最后,我们通过调用Context对象的write方法将最大值写入上下文。
public static class MaxReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
private IntWritable result = new IntWritable();
public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int max = Integer.MIN_VALUE;
for (IntWritable val : values) {
max = Math.max(max, val.get());
}
result.set(max);
context.write(key, result);
}
}
完整的代码如下:
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class MaxValue {
public static class MaxValueMapper extends Mapper<LongWritable, Text, LongWritable, LongWritable> {
private LongWritable lineNumber = new LongWritable();
private LongWritable maxNumber = new LongWritable();
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] numbers = line.split(",");
long max = Long.MIN_VALUE;
for (String number : numbers) {
long currentNumber = Long.parseLong(number.trim());
if (currentNumber > max) {
max = currentNumber;
}
}
lineNumber.set(key.get());
maxNumber.set(max);
context.write(lineNumber, maxNumber);
}
}
public static class MaxValueReducer extends Reducer<LongWritable, LongWritable, LongWritable, LongWritable> {
private LongWritable maxLineNumber = new LongWritable();
private LongWritable maxValue = new LongWritable(Long.MIN_VALUE);
@Override
public void reduce(LongWritable key, Iterable<LongWritable> values, Context context)
throws IOException, InterruptedException {
long localMax = Long.MIN_VALUE;
for (LongWritable value : values) {
long currentValue = value.get();
if (currentValue > localMax) {
localMax = currentValue;
}
}
if (localMax > maxValue.get()) {
maxValue.set(localMax);
maxLineNumber.set(key.get());
}
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
context.write(maxLineNumber, maxValue);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "Max Value");
job.setJarByClass(MaxValue.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setMapperClass(MaxValueMapper.class);
job.setReducerClass(MaxValueReducer.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(LongWritable.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(LongWritable.class);
TextInputFormat.addInputPath(job, new Path(args[0]));
TextOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
创建一个输入文件,其中每行包含一个或多个用逗号分隔的整数,例如:
1, 2, 3, 4, 5
6, 7, 8, 9, 10
11, 12, 13, 14, 15
该文件中包含三行整数,每行包含五个整数。我们的程序将读取该文件,并返回其中的最大值,即15。
接下来,我们可以在Eclipse中创建一个MapReduce项目,并将上述代码复制到项目中的相应文件中。在项目中,我们需要在项目的src/main/resources目录下创建一个名为input的文件夹,并将上述测试数据文件复制到该文件夹中。
现在我们可以运行程序。首先,我们需要将项目打包成JAR文件。在Eclipse中,我们可以右键单击项目,选择Export,然后选择JAR文件并按照向导的指示进行操作。将生成的JAR文件上传到Hadoop集群中,并使用以下命令在Hadoop集群上运行程序:
hadoop jar maxvalue.jar input output
其中,maxvalue.jar是我们生成的JAR文件,input是输入文件夹的路径,output是输出文件夹的路径。运行程序后,Hadoop将在输出文件夹中生成一个名为part-r-00000的文件,其中包含计算出的最大值。我们可以使用以下命令查看该文件中的结果:
hadoop fs -cat output/part-r-00000
该命令将输出键值对(key-value),最终的答案是14 15
更多推荐
所有评论(0)