map reduce原理
问题：怎样解决海量数据的计算？
- MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题.
- MR由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。
执行步骤
map任务处理
- 读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。
- 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
- 对输出的key、value进行分区，
- 对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。
- (可选)分组后的数据进行归约。
reduce任务处理
- 对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。
- 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
- 把reduce的输出保存到文件中
word count实例
image.png
map代码：
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//接收一行数据
String line = value.toString();
//分割
String[] words = line.split(" ");
//迭代
for(String w : words){
//发送
context.write(new Text(w), new LongWritable(1));
}
}
}
reduce代码
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
@Override
protected void reduce(Text key, Iterable<LongWritable> v2s, Context context)
throws IOException, InterruptedException {
//定义一行计数器
long sum = 0;
//迭代他的次数
for(LongWritable lw : v2s){
//求和
sum += lw.get();
}
//输出
context.write(key, new LongWritable(sum));
}
}
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
* 1.定义一个类，这个类是MR执行入口
* 2.定义一个类，这个类要继承org.apache.hadoop.mapreduce.Mapper，指定泛型的类型（k1,v1和k2,v2），然后重新map方法，接收数据，实现具体的业务逻辑
* 3.定义一个类，这个类要继承org.apache.hadoop.mapreduce.Reducer，指定泛型的类型(k2,v2和k3,v3)，然后重新reduce方法，接收数据(k2和v2s，已经分组)，实现具体的业务逻辑
* 4.在main方法中将自定义的mapper和reducer组装起来，提交
* 5.打包
* @author zx
*
*/
public class WordCount {
public static void main(String[] args) throws Exception {
// 构造一个Job对象，job对象要依赖Configuration
Configuration conf = new Configuration();
conf.set("mapreduce.framework.name", "yarn");
conf.set("yarn.resourcemanager.hostname", "itcast01");
Job job = Job.getInstance(conf);
//注意：要将main所在的类设置一下
job.setJarByClass(WordCount.class);
//设置Mapper相关的属性
job.setMapperClass(WCMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));
//设置Reducer相关属性
job.setReducerClass(WCReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setCombinerClass(WCReducer.class);
//提交任务
job.waitForCompletion(true);
}
}
详细代码见：worcount
idea 打jar包
-
选中Java项目工程名称，在菜单中选择 File->project structure...
image.png
-
在弹出的窗口中左侧选中"Artifacts"，点击"+"选择jar，然后选择"from modules with dependencies"。
image.png
-
在配置窗口中配置"Main Class"。
image.png
- 配置“Directory for META-INF/MAINFEST.MF”， 放在resource目录下
- 配置完成后如下图所示，点击OK进入下一步。我个人建议选择“extract to the target JAR”，这样所有依赖的jar包都会放在生成的jar包中。
image.png
- 完成后，点击OK，Apply等按钮，回到IDEA的主菜单，选择“Build - Build Artifacts”下的“Build”或者“Rebuild”即可生成最终的可运行的jar，
上传wordCount.txt到hdfs： hadoop fs -put wordCount.txt
执行MR的命令：
hadoop jar <jar在linux的路径> <main方法所在的类的全类名> <参数>
例子：
hadoop jar mr.jar hdfs://server-2:9000/wordCount.txt /out2
得到结果：
hadoop fs -cat /out2/part-r-00000
hard 1
hello 2
nihao 2
study 1
world 1
wusong 2
yes 1
网友评论