YARN集群的mapreduce测试(五)
将user表计算后的结果分区存储
测试准备:
首先同步时间,然后master先开启hdfs集群,再开启yarn集群;用jps查看:
master上: 先有NameNode、SecondaryNameNode;再有ResourceManager;
slave上: 先有DataNode;再有NodeManager;
如果master启动hdfs和yarn成功,但是slave节点有的不成功,则可以使用如下命令手动启动:
hadoop-daemon.sh start datanode |
yarn-daemon.sh start nodemanager |
然后在集群的主机本地环境创建myinfo.txt;内容如下:
然后将测试文件myinfo.txt上传到集群中:
测试目标:
hadoop集群分区及缓存:
1、分区是必须要经历Shuffle过程的,没有Shuffle过程无法完成分区操作
2、分区是通过MapTask输出的key来完成的,默认的分区算法是数组求模法:
数组求模法:
将Map的输出Key调用hashcode()函数得到的哈希吗(hashcode),此哈希吗是一个数值类型,将此哈希吗数值直接与整数的最大值(Integer.MAXVALUE)取按位与(&)操作,将与操作的结果与ReducerTask
的数量取余数,将此余数作为当前Key落入的Reduce节点的索引;
-------------------------
Integer mod = (Key.hashCode()&Integer.MAXVALUE)%NumReduceTask;
被除数=34567234
NumReduceTas=3
------结果:
0、1、2 这三个数作为Reduce节点的索引;
数组求模法是有HashPartitioner类来实现的,也是MapReduce分区的默认算法;
测试代码:
package com.mmzs.bigdata.yarn.mapreduce; import java.io.IOException; import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper; public class PartitionMapper extends Mapper<LongWritable, Text,LongWritable, Text>{
private LongWritable outKey;
private Text outValue; @Override
protected void setup(Mapper<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
outKey = new LongWritable();
outValue= new Text();
} @Override
protected void cleanup(Mapper<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
outKey=null;
outValue=null;
} @Override
protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
String[] fields=value.toString().split("\\s+");
Long userId=Long.parseLong(fields[0]);
outKey.set(userId);
outValue.set(new StringBuilder(fields[1]).append("\t").append(fields[2]).toString());
context.write(outKey, outValue); } }
PartitionMapper
package com.mmzs.bigdata.yarn.mapreduce; import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Partitioner; public class MyPartitoner extends Partitioner { @Override
public int getPartition(Object key, Object value, int num) {
LongWritable userId=(LongWritable)key;
Long userCode=userId.get();
//分区的依据
if(userCode<6){
return 0;
}else if(userCode<10){
return 1;
}else{
return 2;
}
} }
MyPartitoner
package com.mmzs.bigdata.yarn.mapreduce; import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer; public class PartitionReducer extends Reducer<LongWritable, Text, LongWritable, Text>{ @Override
protected void cleanup(Reducer<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
super.cleanup(context);
} @Override
protected void reduce(LongWritable key, Iterable<Text> values,
Reducer<LongWritable, Text, LongWritable, Text>.Context context) throws IOException, InterruptedException {
Iterator<Text> its= values.iterator();
if(its.hasNext()){
context.write(key, its.next());
}
} @Override
protected void setup(Reducer<LongWritable, Text, LongWritable, Text>.Context context)
throws IOException, InterruptedException {
super.setup(context);
} }
PartitionReducer
package com.mmzs.bigdata.yarn.mapreduce; import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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; public class ParititionDriver {
private static FileSystem fs; private static Configuration conf; static{
String uri="hdfs://master01:9000/";
conf=new Configuration();
try {
fs=FileSystem.get(new URI(uri),conf,"hadoop");
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (URISyntaxException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} public static void main(String[] args) throws Exception{
Job wcJob =getJob(args);
if(null==wcJob)return;
/*
*提交Job到集群并等到Job运行完成,参数true表示将Job的运行是状态信息返回到
*客户端控制台输出,返回值的布尔值代表Job是否运行成功
*/
boolean flag=wcJob.waitForCompletion(true);
System.exit(flag?0:1); }
public static Job getJob(String[] args) throws Exception{
if(null==args||args.length<2)return null;
//放置需要处理的数据所在的HDFS路径
Path inputPath=new Path(args[0]);
//放置Job作业执行完成之后其处理结果的输出路径
Path ouputPath=new Path(args[1]);
//如果输出目录已经存在则将其删除并重建
if(!fs.exists(inputPath))return null;
if(fs.exists(ouputPath)){
fs.delete(ouputPath,true);
}
//获取Job实例
Job wcJob=Job.getInstance(conf, "PartitionerJob");
//设置运行此jar包的入口类
wcJob.setJarByClass(ParititionDriver.class);
//设置job调用的Mapper类
wcJob.setMapperClass(PartitionMapper.class);
//设置job调用的Reducer类(如果一个job没有ReduceTask则此条语句可以不掉用)
wcJob.setReducerClass(PartitionReducer.class);
//设置MapTask的输出值类型
wcJob.setMapOutputKeyClass(LongWritable.class);
//设置MapTask的输出键类型
wcJob.setMapOutputValueClass(Text.class);
//设置整个Job的输出键类型
wcJob.setOutputKeyClass(LongWritable.class);
//设置整个Job的输出值类型
wcJob.setOutputValueClass(Text.class); //设置分区类
wcJob.setPartitionerClass(MyPartitoner.class);
wcJob.setNumReduceTasks(3);//这个数字和MyPartitioner类中的三种分区依据相对应
//如果将数字调整大了,那么只有分区依据的前三个文件有内容,多出任务对应的仅仅是个空分区、空文件;
//如果将数字调整小了,那么将得不到任何一个分区结果 //设置整个Job需要处理数据的输入路径
FileInputFormat.setInputPaths(wcJob, inputPath);
//设置整个Job需要计算结果的输出路径
FileOutputFormat.setOutputPath(wcJob, ouputPath);
return wcJob;
} } ParititionDriver
ParititionDriver
测试结果:
运行时传入参数是:
如果在客户端eclipse上运行:传参需要加上集群的master的uri即 hdfs://master01:9000
输入路径参数: /data/partition/src
输出路径参数: /data/partition/dst
深入测试:(修改PartitionDriver类的如下代码)
//设置分区类
wcJob.setPartitionerClass(MyPartitoner.class);
wcJob.setNumReduceTasks(3);//这个数字和MyPartitioner类中的三种分区依据相对应
//如果将数字调整大了(比如调整为4),那么只有分区依据的前三个文件有内容,多出任务对应的仅仅是个空分区、空文件;
//如果将数字调整小了(比如调整为2),那么将得不到任何一个分区结果
小结:
1、数据量需要达到一定的数量级使用hadoop集群来处理才是划算的
2、集群的计算性能取决于任务数量的多少,设置任务数量必须充分考虑到集群的计算能力(比如:物理节点数量);
a、Map设置的任务数量作为最小值参考
b、Reduce的任务数默认是1(使用的也是默认的Partitioner类),如果设置了则启动设置的数量;
不管MapTask还是ReduceTask,只要任务数量越多则并发能力越强,处理效率会在一定程度上越高,但是设置的任务数量必须参考集群中的物理节点数量,如果设置的任务数量过多,会导致每个物理节点上分摊的任务数量越多,处理器并发每一个任务产生的计算开销越大,任务之间因处理负载导致相互之间的影响非常大,任务失败率上升(任务失败时会重新请求进行计算,最多重新请求3次),计算性能反而下降,因此在设计MapTask与ReduceTask任务数量时必须权衡利弊,折中考虑...
最新文章
- Activity的释放
- Objective-C中的浅拷贝和深拷贝(转载)
- Spring和SpringMVC父子容器关系初窥
- mydumper linux mysql 备份利器
- 关于SQL SERVER的N前缀的理解
- Apache与Tomcat整合
- 中文字体在CSS中的表达方式
- 更新一波题解(最近做的三个dp题)
- osip及eXosip的编译方法
- Filter技术+职责链模式
- Linux驱动技术(四) _异步通知技术
- 用node搭建简单的静态资源管理器
- java8 - IO
- 蜻蜓FM 涉嫌诈骗投资人和广告主源代码剖析
- XenServer 自动化布署 (关键词: PXE ANSWER SCRIPT)
- Lavarel Route::resource
- MQ(2)---JMS
- java.sql.SQLException: Access denied for user &#39;scott&#39;@&#39;localhost&#39; (using password: YES)
- 设计 MySQL 数据表的时候一般都有一列为自增 ID,这样设计原因是什么,有什么好处?
- 使用btrace需要注意的几个问题