MapReduce和yarn集群

 背景：我们在对HDFS上的文件进行操作的时候，之前我们需要通过先将原来的文件进行下载，在本地上进行编码运行和调试，再改文件并将其传入到hdfs中，但是这样的操作给我们带来了很多的不便。以下介绍通过框架MapReduce和yarn集群对HDFS上的数据进行操作。

 MapReduce是一种编程模型，用于大规模数据集（大于1TB）的并行运算。概念"Map（映射）“和"Reduce（归约）”，是它们的主要思想，都是从函数式编程语言里借来的，还有从矢量编程语言里借来的特性。它极大地方便了编程人员在不会分布式并行编程的情况下，将自己的程序运行在分布式系统上。 当前的软件实现是指定一个Map（映射）函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（归约）函数，用来保证所有映射的键值对中的每一个共享相同的键组。 （摘自百度百科）

MapReduce主要分为两个阶段：

map阶段——程序MapTaskreduce阶段——程序ReduceTsak

yarn集群的工作调度机制

实验内容，对hdfs中的的一个txt文本进行英文单词的统计。

第一种方法

 这里的第一种方法是将我们的程序完全的放在windows下运行，不过在运行的时候我们要将文件通过yarn集群传递到hdfs（虚拟机）里面就会出很多的问题，例如：上传时的用户名设置、文件路径设置、jar包的设置等等。

程序逻辑图

第一步：导包

导入我们需要的jar包，这里我们就是用maven来导入我们需要的jar包

<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.8.1</version> </dependency>

第二步：编写Mapper类

这个类是将我们在HDFS中获取到的文件一行一行的读入，并将格式进行切割数据，其按照key，value值进行分类，并将分类好的数据保存到contxt中，传递给Reducer。

import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; /* * 在Mapper<>中 * 第一个参数:是map task读取到的数据的key的类型，是一行的起始偏移量Long * 第二个参数：是map task读取到的数据的value的类型，是一行的内容String * 第三个参数：用户的自定义map方法要返回的结果kv数据的key的类型，在wordcount逻辑中，我们需要返回的是单词String * 第四个参数：用户的自定义map方法要返回的结果kv数据的value的类型，在wordcount逻辑中，我们需要返回的是单词整数Integer * * 但是在MapReduce中，map产生的数据需要传输给reduce，需要进行序列化和反序列化，而jdk中的原生序列化机制产生的数据量比较冗余，就会导致数据在MapReduce在运行过程中传输效率低 * 所以hadoop专门设计了自己的序列化机制，那么，MapReduce中传输的数据类型就必须实现hadoop自己的序列化接口 * * hadoop为jdk中的常用基本数据类型Long String Integer Float等数据类型封装了自己的实现了hadoop序列化的接口类型： * Long—>LongWritable * String->Text * Integer->IntWritable * Float->FloatWritable * */ public class WordcountMapper extends Mapper<LongWritable, Text,Text, IntWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 切单词 String line = value.toString(); String[] words = line.split(" "); for (String word : words) { // context.write(word,1);这样写会报错，因为context的两个参数的类型是上面我们封装的Text和Inwritable的类型，所以在输入String和int类型时会有问题 context.write(new Text(word),new IntWritable(1)); } } }

但是我们这里要注意，在这里面的字符串类型都要写成我们在hadoop给我们封装的字符串类型，

hadoop为jdk中的常用基本数据类型Long String Integer Float等数据类型封装了自己的实现了hadoop序列化的接口类型：

Long—>LongWritableString->TextInteger->IntWritableFloat->FloatWritable

第三步：编写Reducer类

 这个类就是对Mapper这个类传过来的（Key，Value）值进行数据上的操作，并将其返回给任务区写回到虚拟机中。

import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; /* * 第一个参数：对应的应该是我们在WordcountMapper里面传给我们的第三个参数 * 第二个参数：对应的应该是我们在WordcountMapper里面传个我们的第四个参数 * 第三个参数：对应的应该是我们的单词，String类型，所以应该是Text * 第四个参数：对应的应该是我们的总数，int类型，所以应该是InWritable * */ public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException { int count = 0; Iterator<IntWritable> iterator = values.iterator(); while(iterator.hasNext()){ IntWritable value = iterator.next(); count += value.get(); } context.write(key, new IntWritable(count)); } }

第四步：在虚拟机中yarn的程序

yarn集群中有两个角色：

主节点：Resource Manager 1台

从节点：Node Manager N台

Resource Manager一般安装在一台专门的机器上

Node Manager应该与HDFS中的data node重叠在一起

修改namenode虚拟机中yarn-site.xml

<property> <name>yarn.resourcemanager.hostname</name> <value>hdp-04</value> <!--写自己虚拟机的名称--> </property> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> <property> <name>yarn.nodemanager.resource.memory-mb</name> <value>2048</value> <!--给每一个任务的内存量--> </property> <property> <name>yarn.nodemanager.resource.cpu-vcores</name> <value>2</value> </property>

然后复制到每一台机器上

然后在第一台虚拟机上，修改hadoop的slaves文件，列入要启动nodemanager的机器

然后将到所有机器的免密登陆配置好

然后，就可以用脚本启动yarn集群：

sbin/start-yarn.sh

停止：

sbin/stop-yarn.sh

第五步：编写主函数类

主函数类的主要的目的：

封装本次job运行时所需要的必要函数

跟yarn进行怀虎，将MapReduce程序成功的启动，运行

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; import java.net.URI; public class JobSubmitter { public static void main(String[] args) throws Exception { // 在代码中设置JVM系统参数，用于给job对象来获取访问HDFS的用户身份 System.setProperty("HADOOP_USER_NAME","kjxy"); Configuration conf=new Configuration(); // 1.设置job运行时要访问的默认文件系统 conf.set("fs.defaultFS","hdfs://192.168.200.137:9000"); // 2.设置job提交到哪里去运行 conf.set("mapreduce.framework.name","yarn"); conf.set("yarn.resourcemanager.hostname","192.168.200.137"); // 3、如果从windows系统上运行这个job提交客户端程序，则需要加这个跨平台提交的参数 conf.set("mapreduce.app-submission.cross-platform","true"); Job job = Job.getInstance(conf); // 1、封装参数：jar包所在的位置,获取到本类的路径 job.setJar("d:/wc.jar"); // job.setJarByClass(JobSubmitter.class); // 2、封装参数：本次job所要调用的Mapper实现类、Reducer实现类 job.setMapperClass(WordcountMapper.class); job.setReducerClass(WordcountReducer.class); // 3、封装参数：本次job的Mapper实现类产生的结果数据的key，value类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); Path output=new Path("/wordcount/output"); FileSystem fs=FileSystem.get(new URI("hdfs://192.168.200.137:9000"),conf,"kjxy"); if(fs.exists(output)){ fs.delete(output,true); } // 4、封装参数：本次job要处理的输入数据集所在路径 FileInputFormat.setInputPaths(job,new Path("/wordcount/input/")); // 5、最终结果的输出路径 FileOutputFormat.setOutputPath(job,output);//注意输出路径必须不存在，如果已存在会抛异常的 // 6、封装参数：想要启动的reduce task的数量 job.setNumReduceTasks(2); // 7、提交job给yarn boolean res = job.waitForCompletion(true); System.exit(res?0:1); } }