经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » 大数据/云/AI » Hadoop » 查看文章
Hadoop学习(3)-mapreduce快速入门加yarn的安装
来源:cnblogs  作者:两千个秘密  时间:2019/7/25 8:21:28  对本文有异议

mapreduce是一个运算框架,让多台机器进行并行进行运算,

他把所有的计算都分为两个阶段,一个是map阶段,一个是reduce阶段

 

map阶段:读取hdfs中的文件,分给多个机器上的maptask,分文件的时候是按照文件的大小分的

比如每个maptask都会处理128M的文件大小,然后有个500M的文件,就会启动ceil(500/128)个maptask

每读取文件的一行的处理,需要自己去写,注意每个maptask的处理逻辑都是一样的

处理出来的结果一定是一对key和value。

maptask里面的方法叫map(long k, string v, context);   k是文件的起始偏移量,v是内容,

context是要把产生的key,value对放入的容器。

 

 

 

reduce阶段:每个机器上有reducetask,其作用是对maptask产生的key和value进行聚合

聚合的原则是key一样的一定分发给一个reducetask,这个操作叫做shuffle

然后把相同key的数据作为一组进行处理。最后会把结果写到hdfs里面

每有几个reducetask,就会生成几个part-r-xxxx文件

reducetask里面的方法reduce(k,value迭代器,context),k就是key,迭代器遍历每一个key相同的value,然后context就是写入hdfs里的,也是一个key和value

 

入门样例:wordcount

设计思路,每个maptask读取文件,

map里面k  起始偏移量没用,我们每读一行v,产生就是key是每一个单词,然后value就定为1就行,把这个key,value放入context里面

在reduce阶段,每个key相同的就会作为一组,也就是单词相同的作为一组,就统计出现几次就行。

 

开始在esclipe写mapreduce的业务逻辑,首先我们需要一些jar包,相关的jar包在解压出来的hadoop下的share/hadoop文件夹下

这几个文件夹下的jar包和这几个文件夹下的lib下的jar包都拷贝到esclipe再buildPath

首先编写mapper方法

  1. package test;
  2. import java.io.IOException;
  3. import org.apache.hadoop.io.IntWritable;
  4. import org.apache.hadoop.io.LongWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Mapper;
  7. /**
  8. * KEYIN :是map task读取到的数据的key的类型,是一行的起始偏移量Long
  9. * VALUEIN:是map task读取到的数据的value的类型,是一行的内容String
  10. *
  11. * KEYOUT:是用户的自定义map方法要返回的结果kv数据的key的类型,在wordcount逻辑中,我们需要返回的是单词String
  12. * VALUEOUT:是用户的自定义map方法要返回的结果kv数据的value的类型,在wordcount逻辑中,我们需要返回的是整数Integer
  13. *
  14. *
  15. * 但是,在mapreduce中,map产生的数据需要传输给reduce,需要进行序列化和反序列化,而jdk中的原生序列化机制产生的数据量比较冗余,就会导致数据在mapreduce运行过程中传输效率低下
  16. * 所以,hadoop专门设计了自己的序列化机制,那么,mapreduce中传输的数据类型就必须实现hadoop自己的序列化接口
  17. *
  18. * hadoop为jdk中的常用基本类型Long String Integer Float等数据类型封住了自己的实现了hadoop序列化接口的类型:LongWritable,Text,IntWritable,FloatWritable
  19. *
  20. *
  21. *
  22. *
  23. */
  24. //第一个泛型为起始偏移量,没啥用,第二个为字符串,为读取到的一行内容,第三个,第四个为context中的key,和value,即发送给reduce的k,v对
  25. public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
  26. @Override
  27. //重写map方法
  28. protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  29. // 切单词
  30. String line = value.toString();
  31. String[] words = line.split(" ");
  32. for(String word:words){
  33. context.write(new Text(word), new IntWritable(1));
  34. }
  35. }
  36. }

接下来是reduce类

  1. package test;
  2. import java.io.IOException;
  3. import java.util.Iterator;
  4. import org.apache.hadoop.io.IntWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Reducer;
  7. //第一个,第二个为接收到的map的key,value,第三第四为写入到hdfs的key,value
  8. public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
  9. @Override
  10. //一个key,众多value的迭代器,一个context;
  11. protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
  12. int count = 0;
  13. Iterator<IntWritable> iterator = values.iterator();
  14. while(iterator.hasNext()){
  15. IntWritable value = iterator.next();
  16. count += value.get();
  17. }
  18. context.write(key, new IntWritable(count));
  19. }
  20. }

 

 

 然而我们写的程序需要提交给我们的hadoop集群去运行,而管理这个事情的就是我们的yarn

yarn是一个分布式程序的运行调度平台

yarn中有两大核心角色

1、Resource Manager

接受用户提交的分布式计算程序,并为其划分资源

接收客户端要运行几个容器,进行任务调度

管理、监控各个Node Manager上的资源情况,以便于均衡负载

2、Node Manager

 

管理它所在机器的运算资源创建容器(cpu + 内存)

负责接受Resource Manager分配的任务,创建容器、回收资源

我们需要把我们的程序的jar包分发给每一个NodeManager,让他们去运行,

 

node manager在物理上应该跟data node部署在一起

resource manager在物理上应该独立部署在一台专门的机器上

 

yarn的安装

yarn我们不需要再下载了,在我们的hadoop里面已经有了yarn,我们只需要写一下配置文件就行

[root@hdp-04 ~]# vi apps/hadoop-2.8.1/etc/hadoop/yarn-site.xml

第一个指明哪一台机器当做resourcemanager,第二个指明nodemanager的任务是什么

 

  1. <configuration>
  2.  
  3. <!-- Site specific YARN configuration properties -->
  4. <property>
  5. <name>yarn.resourcemanager.hostname</name>
  6. <value>hdp-01</value>
  7. </property>
  8.  
  9. <property>
  10. <name>yarn.nodemanager.aux-services</name>
  11. <value>mapreduce_shuffle</value>
  12. </property>
  13.  
  14. </configuration>

然后复制这个文件到你的其他机器上

在你的resourcemanager机器上敲 start-yarn.sh,(关闭时stop-yarn.sh)

hadoop就会启动resourcemanager,其他的nodemanager,hadoop是通过slave文件知道的(在/root/apps/hadoop-2.8.1/etc/hadoop/slaves),里面写入你的nodemanager的ip就行,一行一个。

启动之后可以敲jps看一下

或者看网页的形式,resourcemanager的端口号是8088.比如hdp-01:8088

 

然后安装完yarn之后嘞,我们就可以写一个java的提交任务的程序了

 

  1. package test;
  2. import java.net.URI;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.FileSystem;
  5. import org.apache.hadoop.fs.Path;
  6. import org.apache.hadoop.io.IntWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  10. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  11. public class JobSubmitter {
  12. public static void main(String[] args) throws Exception {
  13. //在JVM中设置访问hdfs的用户身份为root,因为要对存在datanode节点的文件进行读写,不然可能会权限不够
  14. // 构造一个访问指定HDFS系统的客户端对象: 参数1:——HDFS系统的URI,参数2:——客户端要特别指定的参数,参数3:客户端的身份(用户名)
  15. //FileSystem fs = FileSystem.get(new URI("hdfs://172.31.2.38:9000/"), conf, "root");
  16. //如果是这样设置访问用户身份是不行的,因为不光是自己的客户端访问hdfs,
  17. //job还会创建自己的hdfs的对象FileSystem去访问datanode,那么job创建的对象是从系统环境变量拿到的用户名,所以这样设置身份
  18. System.setProperty("HADOOP_USER_NAME", "root");
  19. //设置配置参数
  20. Configuration conf = new Configuration();
  21. //设置job运行时要访问的默认文件系统
  22. conf.set("fs.defaultFS", "hdfs://172.31.2.38:9000");
  23. //设置job要提交到那里去运行,可以是yarn,也可以是local
  24. conf.set("mapreduce.framework.name", "yarn");
  25. //设置resourcemanager在哪
  26. conf.set("yarn.resourcemanager.hostname", "172.31.2.38");
  27. //如果从windows提交job,需要设置跨平台提交时,把windows中的命令,替换成linux的
  28. //比如运行jar包中某个程序,在linux和windows是不一样的,这样可以自动转化
  29. conf.set("mapreduce.app-submission.cross-platform","true");
  30. //设置job
  31. Job job = Job.getInstance(conf);
  32. //封装jar包在windows下的位置
  33. job.setJar("d:/wc.jar");
  34. //设置本次job所要调用的Mapper的class类和reduce的class类
  35. job.setMapperClass(WordcountMapper.class);
  36. job.setReducerClass(WordcountReducer.class);
  37. //设置mapper实现类的产生结果的key,value类型
  38. job.setMapOutputKeyClass(Text.class);
  39. job.setMapOutputValueClass(IntWritable.class);
  40. //设置reduce实现类的产生结果的key,value类型
  41. job.setOutputKeyClass(Text.class);
  42. job.setOutputValueClass(IntWritable.class);
  43. //设置map时,job要处理的数据的路径,和产生的结果的路径在哪
  44. FileInputFormat.setInputPaths(job, new Path("/wordcount/input"));
  45. //注意输出路径一定要不存在
  46. FileOutputFormat.setOutputPath(job, new Path("/wordcount/output"));
  47. //设置想要启动reduce task的数量是多少
  48. job.setNumReduceTasks(2);
  49. //提交给yarn,等待这个job完成才退出
  50. boolean res = job.waitForCompletion(true);
  51. System.exit(res?0:-1);
  52. }
  53. }

 额外知识点:

maven创建报错说插件下载失败,右键项目

然后 右键属性Maven,Update project

把这个给点上就行了。或者进入到org/apache/maven,把里面的东西全删了,让他自己去下载。

 

在编程时的易错点:

注意自己写的路径file:\为windows,/为linux

注意改完源码之后,要注意重新生成一个jar包,不然提交到linux里,还是会报错

 

原文链接:http://www.cnblogs.com/wpbing/p/11238504.html

 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号