1. MapReduce 统计文件的单词出现的个数
Mapper: 处理具体文本，发送结果
Reducer: 合并各个Mapper发送过来的结果
Job: 制定相关配置，框架
Mapper
package cn.itcast.hadoop.mr.wordcount;import java.io.IOException;import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.util.StringUtils;// 4个泛型中，前两个是指定mapper输入数据的类型 // map和 reducer的输入和输出都是key-value对的形式 // 默认情况下。框架输入的我们mapper的输入数据中，key是要处理的文本中的一行的起始偏移量， 内容就是value public class WCMapper extends Mapper <LongWritable, Text, Text, LongWritable> {// 每读一次数据，就调一次这个方法 @Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{//具体业务逻辑就像和在这里， 传入数据就是 key， value//将这一行的内容转化成string类型String line = value.toString();// 对这一行的文本按特定分隔符切分String[] words = StringUtils.split(line, ' ');// 便利这个单词数组， 输出为kv形式 k：单词 v：1for (String word : words) {context.write(new Text(word), new LongWritable(1));}} }
Reducer
package cn.itcast.hadoop.mr.wordcount;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> {@Overrideprotected void reduce(Text Key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {long count = 0;for (LongWritable value:values) {count += value.get();}context.write(Key, new LongWritable(count));} }
Job
package cn.itcast.hadoop.mr.wordcount;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.io.Text; //import com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider.Text;public class WCRunner {public static void main(String[] args) throws Exception{Configuration conf = new Configuration();Job wcJob = Job.getInstance(conf);wcJob.setJarByClass(WCRunner.class);//本job使用的mapper和reducer类wcJob.setMapperClass(WCMapper.class);wcJob.setReducerClass(WCReducer.class);// 指定reduce的输出数据kv类型wcJob.setOutputKeyClass(Text.class);wcJob.setOutputValueClass(LongWritable.class);// 指定mapper的输出数据kv类型wcJob.setMapOutputKeyClass(Text.class);wcJob.setMapOutputValueClass(LongWritable.class);//指定要处理的输入数据存放路径FileInputFormat.setInputPaths(wcJob, new Path("/wc/srcdata/"));//指定处理结果的输出数据存放路径FileOutputFormat.setOutputPath(wcJob, new Path("/wc/output"));//将job提交给集群运行wcJob.waitForCompletion(true);} }
2. Yarn资源调度框架
Resource Manager:
Node Manager:
1. wcJob.waitforCompleition启动一个RunJar进程，这个进程向RM申请执行一个Job 2. RM 返回一个Job相关资源的路径staging-dir，和为Job产生的jobID 3. RunJar提交资源到 HDFS的 staging-dir上 4. RunJar提交资源完毕之后，上报RM 提交资源完毕 5. RM下个Job加入RM中的任务队列中 6. 各个Node Manager通过通信，从RM的任务队列中领取任务 7. 各个Node Manager初始化 运行资源的容器，从staging-dir上面拉取资源 8. RM选择一个Node Manager 启动MRAppMaster 来运行map reducer 9. MRAppMaster向RM注册 10. MSAppMaster启动Mapper任务 11. MSAppMaster启动Reducer任务 12. 任务完成后， 向RM注销自己
3.几种运行模式
本地模型运行
在windows的eclipse里面直接运行main方法，就会将job提交给本地执行器locaJobRunner执行
– 输入输出数据可以放在本地路径下（c:/wc/src/data/）
– 输入输出数据也可以放在hdfs中(hdfs://hadoop1:9000/wc/srcdata)
在linux的ecllipse里面直接运行main 方法，则不需要添加yarn相关的配置，也会提交给localJobRunner执行
– 输入输出数据可以放在本地路径下（/home/hadoop/wc/srcdata）
– 输入输出数据也可以放在hdfs中(hdfs://hadoop1:9000/wc/srcdata)
集群模式运行
将工程打包成jar包， 上传到服务器，然后用hadoop命令提交 hadoop jar wc.jar cn.itcast.hadoop.mr.wordcount.WCRunner
在linux的eclipse中直接运行main方法，也可以提交到集群中去运行，但是，必须采取一下措施：
– 在工程src目录下加入mapred-site.xml 和yarn-site.xml
– 将工程打成jar包（wc.jar), 同时在main方法中添加一个conf的配置参数 conf.set(“mapreduce.job.jar”, “wc.jar”);
在windows的eclipse中直接运行main方法，也可以提交集群中运行，但是因为平台不兼容，需要做很多的设置修改
– 要在windows中存放一份hadoop的安装包（解压好的）
– 要将其中的lib和bin目录替换成根据你的windows版本重新编译出的文件
– 再要配置系统环境变量 HDOOP_HOME 和 PATH
– 修改YarnRunner这个类的源码
