之前学习了一段时间的hadoop的相关知识 ,学习理论基础的时候要同时实际 *** 作才能对它更熟练,废话不多说来说说在hadoop上运行一个最简单的words count的程序
首先我先贴上这个程序的源代码 供大家参考 代码分为三个部分写的Run、 map阶段、 reduce阶段
Map:
[java] view plain copy
<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;
import javaioIOException;
import javautilStringTokenizer;
import orgapachehadoopioIntWritable;
import orgapachehadoopioLongWritable;
import orgapachehadoopioText;
import orgapachehadoopmapreduceMapper;
public class WordsMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>Context context)
throws IOException, InterruptedException {
String line = valuetoString();
StringTokenizer st = new StringTokenizer(line);
while(sthasMoreTokens()){
String word = stnextToken();
contextwrite(new Text(word), new IntWritable(1));
}
}
}</span>
Reduce:
[java] view plain copy
<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;
import javaioIOException;
import orgapachehadoopioIntWritable;
import orgapachehadoopioText;
import orgapachehadoopmapreduceReducer;
public class WordsReduce extends Reducer<Text, IntWritable, Text, IntWritable>{
@Override
protected void reduce(Text key, Iterable<IntWritable> iterator,
Reducer<Text, IntWritable, Text, IntWritable>Context context) throws IOException, InterruptedException {
// TODO 自动生成的方法存根
int sum = 0;
for(IntWritable i:iterator){
sum = sum + iget();
}
contextwrite(key, new IntWritable(sum));
}
}</span>
Run:
[java] view plain copy
<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;
import orgapachehadoopconfConfiguration;
import orgapachehadoopfsPath;
import orgapachehadoopioIntWritable;
import orgapachehadoopioText;
import orgapachehadoopmapreduceJob;
import orgapachehadoopmapreducelibinputFileInputFormat;
import orgapachehadoopmapreducelibinputTextInputFormat;
import orgapachehadoopmapreduceliboutputFileOutputFormat;
import orgapachehadoopmapreduceliboutputTextOutputFormat;
public class Run {
public static void main(String[] args) throws Exception{
// TODO 自动生成的方法存根
Configuration configuration = new Configuration();
Job job = new Job(configuration);
jobsetJarByClass(Runclass);
jobsetJobName("words count!");
jobsetOutputKeyClass(Textclass);
jobsetOutputValueClass(IntWritableclass);
jobsetInputFormatClass(TextInputFormatclass);
jobsetOutputFormatClass(TextOutputFormatclass);
jobsetMapperClass(WordsMapperclass);
jobsetReducerClass(WordsReduceclass);
FileInputFormataddInputPath(job, new Path("hdfs://1921681111:9000/user/input/wc/"));
FileOutputFormatsetOutputPath(job,new Path("hdfs://1921681111:9000/user/result/"));
jobwaitForCompletion(true);
}
}</span>
Run里面的输入和输出路径根据自己的来修改
这个程序就不用去讲解了吧 到处都能找到
首先在hadoop上运行这个程序用两个方法
方法一:将自己的编译软件与hadoop相连(我用的是MyEclipse去链接hadoop),直接运行程序。MyEclipse连接hadoop的教程待会我会在文章结尾处给出一个链接供大家参考。
看到下面的信息就表示你成功了 然后你在再到你的输出文件夹里面就能查看运行的结果了
第二个文件里面的内容就是输出结果
第二种方法:将mapreduce程序打包成jar文件
这里简单的说一下打包的方法
然后下一步,完成就可以了
将打包好的jar文件传到你的装hadoop的机器上(我的hadoop集群是装在linux虚拟机中的)用SSH把jar传过去之后:
在你安装hadoop的目录下的bin目录下有一个hadoop的可执行文件,然后执行下面的 *** 作就可以了:
来解释下我的shell语句
/home/xiaohuihui/wordscountjar:打包之后的jar文件的所在位置(传到虚拟机中位置)
wordsCount/Run:这个位你的jar包中的主函数(这里的主函数就是Runclass)的名字 可以打开你的jar文件查看便知道
还可以在这个语句之后加上你的输入和输出的文件路径,但是这个我已经在我的程序中设置了
如果你运行上面的shell语句之后看到下面的输出,那恭喜你,成功了!!
查看结果你可通过在你的Eclipse连接好hadoop查看,还可以通过在hdfs文件系统的网页去查看(localhost:50070)。
还有一个很重要的一步就是,运行之前保证你的hadoop已经启动了,可以通过jps来查看你的进程中是否已经启动hadoop集群
在安装Hadoop集群的时候,我们在yarn-sitexml文件中配置了MapReduce的运行方式为yarnnodemanageraux-services=mapreduce_shuffle。本节就来详细介绍一下MapReduce的shuffle过程。
shuffle,即混洗、洗牌的意思,是指MapReduce程序在执行过程中,数据在各个Mapper(Combiner、Sorter、Partitioner)、Reducer等进程之间互相交换的过程。
关于上图Shuffle过程的几点说明:
说明:map节点执行map task任务生成map的输出结果。
shuffle的工作内容:
从运算效率的出发点,map输出结果优先存储在map节点的内存中。每个map task都有一个内存缓冲区,存储着map的输出结果,当达到内存缓冲区的阀值(80%)时,需要将缓冲区中的数据以一个临时文件的方式存到磁盘,当整个map task结束后再对磁盘中这个map task所产生的所有临时文件做合并,生成最终的输出文件。最后,等待reduce task来拉取数据。当然,如果map task的结果不大,能够完全存储到内存缓冲区,且未达到内存缓冲区的阀值,那么就不会有写临时文件到磁盘的 *** 作,也不会有后面的合并。
详细过程如下:
(1)map task任务执行,输入数据的来源是:HDFS的block。当然在mapreduce概念中,map task读取的是split分片。split与block的对应关系:一对一(默认)。
此处有必要说明一下block与split:
block(物理划分):文件上传到HDFS,就要划分数据成块,这里的划分属于物理的划分,块的大小可配置(默认:第一代为64M,第二代为128M)可通过 dfsblocksize配置。为保证数据的安 全,block采用冗余机制:默认为3份,可通过dfsreplication配置。注意:当更改块大小的配置后,新上传的文件的块大小为新配置的值,以前上传的文件的块大小为以前的配置值。
split(逻辑划分):Hadoop中split划分属于逻辑上的划分,目的只是为了让map task更好地获取数据。split是通过hadoop中的InputFormat接口中的getSplit()方法得到的。那么,split的大小具体怎么得到呢?
首先介绍几个数据量:
totalSize:整个mapreduce job所有输入的总大小。注意:基本单位是block个数,而不是Bytes个数。
numSplits:来自jobgetNumMapTasks(),即在job启动时用户利用 orgapachehadoopmapredJobConfsetNumMapTasks(int n)设置的值,从方法的名称上看,是用于设置map的个数。但是,最终map的个数也就是split的个数并不一定取用户设置的这个值,用户设置的map个数值只是给最终的map个数一个提示,只是一个影响因素,而不是决定因素。
goalSize:totalSize/numSplits,即期望的split的大小,也就是每个mapper处理多少的数据。但是仅仅是期望
minSize:split的最小值,该值可由两个途径设置:
最终取goalSize和minSize中的最大值!
最终:split大小的计算原则:finalSplitSize=max(minSize,min(goalSize,blockSize))
那么,map的个数=totalSize/finalSplitSize
注意: 新版的API中InputSplit划分算法不再考虑用户设定的Map Task个数,而是用mapredmaxsplitsize(记为maxSize)代替
即:InputSplit大小的计算公式为:splitSize=max{minSize,min{maxSize,blockSize}}
接下来就简答说说怎么根据业务需求,调整map的个数。
当我们用hadoop处理大批量的大数据时,一种最常见的情况就是job启动的mapper数量太多而超出系统限制,导致hadoop抛出异常终止执行。
解决方案:减少mapper的数量!具体如下:
a输入文件数量巨大,但不是小文件
这种情况可通过增大每个mapper的inputsize,即增大minSize或者增大blockSize来减少所需的mapper的数量。增大blocksize通常不可行,因为HDFS被hadoop namenode -format之后,blocksize就已经确定了(由格式化时dfsblocksize决定),如果要更改blocksize,需要重新格式化HDFS,这样当然会丢失已有的数据。所以通常情况下只能增大minSize,即增大mapredminsplitsize的值。
b输入文件数量巨大,且都是小文件
所谓小文件,就是单个文件的size小于blockSize。这种情况通过增大mapredminsplitsize不可行,需要使用FileInputFormat衍生的CombineFileInputFormat将多个input path合并成一个InputSplit送给mapper处理,从而减少mapper的数量。增加mapper的数量,可以通过减少每个mapper的输入做到,即减小blockSize或者减少mapredminsplitsize的值。
(2)map执行后,得到key/value键值对。接下来的问题就是,这些键值对应该交给哪个reduce做?注意:reduce的个数是允许用户在提交job时,通过设置方法设置的!
MapReduce提供partitioner接口解决上述问题。默认 *** 作是:对key hash后再以reduce task数量取模,返回值决定着该键值对应该由哪个reduce处理。这种默认的取模方式只是为了平均reduce的处理能力,防止数据倾斜,保证负载均衡。如果用户自己对Partition有需求,可以自行定制并设置到job上。
接下来,需要将key/value以及Partition结果都写入到缓冲区,缓冲区的作用:批量收集map结果,减少磁盘IO的影响。当然,写入之前,这些数据都会被序列化成字节数组。而整个内存缓冲区就是一个字节数组。这个内存缓冲区是有大小限制的,默认100MB。当map task的输出结果很多时,就可能撑爆内存。需将缓冲区的数据临时写入磁盘,然后重新利用这块缓冲区。
从内存往磁盘写数据被称为Spill(溢写),由单独线程完成,不影响往缓冲区写map结果的线程。溢写比例:spillpercent(默认08)。
当缓冲区的数据达到阀值,溢写线程启动,锁定这80MB的内存,执行溢写过程。剩下的20MB继续写入map task的输出结果。互不干涉!
当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是mapreduce模型的默认行为,也是对序列化的字节做的排序。排序规则:字典排序!
map task的输出结果写入内存后,当溢写线程未启动时,对输出结果并没有做任何的合并。从官方图可以看出,合并是体现在溢写的临时磁盘文件上的,且这种合并是对不同的reduce端的数值做的合并。所以溢写过程一个很重要的细节在于,如果有很多个key/value对需要发送到某个reduce端,那么需要将这些键值对拼接到一块,减少与partition相关的索引记录。如果client设置Combiner,其会将有相同key的key/value对的value加起来,减少溢写到磁盘的数据量。注意:这里的合并并不能保证map结果中所有的相同的key值的键值对的value都合并了,它合并的范围只是这80MB,它能保证的是在每个单独的溢写文件中所有键值对的key值均不相同!
溢写生成的临时文件的个数随着map输出结果的数据量变大而增多,当整个map task完成,内存中的数据也全部溢写到磁盘的一个溢写文件。也就是说,不论任何情况下,溢写过程生成的溢写文件至少有一个!但是最终的文件只能有一个,需要将这些溢写文件归并到一起,称为merge。merge是将所有的溢写文件归并到一个文件,结合上面所描述的combiner的作用范围,归并得到的文件内键值对有可能拥有相同的key,这个过程如果client设置过Combiner,也会合并相同的key值的键值对,如果没有,merge得到的就是键值集合,如{“aaa”, [5, 8, 2, …]}。注意:combiner的合理设置可以提高效率,但是如果使用不当会影响效率!
至此,map端的所有工作都已经结束!
当mapreduce任务提交后,reduce task就不断通过RPC从JobTracker那里获取map task是否完成的信息,如果获知某台TaskTracker上的map task执行完成,Shuffle的后半段过程就开始启动。其实呢,reduce task在执行之前的工作就是:不断地拉取当前job里每个map task的最终结果,并对不同地方拉取过来的数据不断地做merge,也最终形成一个文件作为reduce task的输入文件。
1Copy过程,简单地拉取数据。Reduce进程启动一些数据copy线程(Fether),通过>
以上就是关于怎样在win 7的eclipse上运行mapreduce全部的内容,包括:怎样在win 7的eclipse上运行mapreduce、MapReduce执行过程、如何在hadoop环境下执行mapreduce任务等相关内容解答,如果想了解更多相关内容,可以关注我们,你们的支持是我们更新的动力!
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)