11.11大促主会场
新人页面
精选商品
首月0月租体验,领12个月京东PLUS
自营热卖

014 Shuffle的概念介绍 Shuffle的细节图描述 分区案例 倒排索引案例

初恋的背后 1年前   阅读数 162 0

MapReduce的shuffle过程

在这里插入图片描述
map输出 reduce输入 的过程

在这里插入图片描述
在这里插入图片描述
好理解吧
输入 三行
做分片 一个分片一行
Mapping统计好
Shuffling这边排序 做一些分区的工作
Reducing接收 统计 输出结果
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
这些图讲的都是shuffle
不太清楚
可自行到官网上寻找

然后这里一张长图 拆开了
一块是128M
这里我们默认一块对应一个分片
但是有10%的slop冗余
比如说有140M的文件 也会分成一个分片
一个分片对应一个MapTask
在这里插入图片描述
这个圆形缓冲区默认大小100M
80M是阈值
当达到80M时 就将数据写入到临时文件取
剩下的20M继续接受map函数发送的数据
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
然后可以看一下FileInputFormat和 FileOutputFormat以及SpillRecord的源码 自己分析一下


shuffle的相关操作

在这里插入图片描述

看下配置名

cd /usr/local/hadoop-2.7.1
vi ./etc/hadoop/core-site.xml

在这里插入图片描述
在这里插入图片描述


遇到错误

中间我隔了一天继续这个
再次启动start-all.sh
这个命令后
出现错误
使用hdfs dfs的相关命令出错
在这里插入图片描述
表示就是两个namenode全是standby状态

没有一个active的
难受
网上查阅资料
说是可以手动开启
或者启动zkfc 说是这个没开 开了就好了
所以我就使用了

hadoop-damons.sh start zkfc

这个命令
jps后
在这里插入图片描述
Zk这个服务出现了
没使用上述命令前是没有的 onMygod 难受


我需要上传par文件 就是数据
现在home下建一个par然后写入 复制上就行
然后传到hdfs
然后将jar包传到home下
(导出的jar包可以是整个项目 使用时写明白类名就行)
具体的把 看截图吧

hdfs dfs -put /home/par /par

在这里插入图片描述
在这里插入图片描述
然后记得看下输出目录(我们自己定义) 我们经常使用hdfs下的out目录 注意看看out下的名字 别重复了
我刚刚删了 重新建的 所以我就别看了
总之就是输出别重复了名字

先进入hadoop目录

cd /usr/local/hadoop-2.7.1

然后

yarn jar /home/wc.jar qf.com.mr.PartitionDemo /par /out/19

这个又出错了

上次的错误
连接不上02
在这里插入图片描述
然后
我又

stop-all.sh
然后等着关了后
又
start-all.sh

就是又重启了服务就好了

我猜
是不是
先将zkfc启动了
再去启动其他的服务
是不是这样?

谁能告诉我
找到答案了

在这里插入图片描述

这样再运行就对了
因为数据小
一个分片 一块 就是一个进程
然后四个结果文件
如下图

在这里插入图片描述
然后查看一下 19文件夹下确实有4个结果文件(除了代表成功的那个)
在这里插入图片描述
查看一下其中内容
非常正确
在这里插入图片描述
至此
分区案例结束

下面贴上代码
写的时候注意导入的包

PartitionDemo

package qf.com.mr;

import java.io.IOException;


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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


/* *@author Shishuai E-mail:1198319583@qq.com *@version Create time : 2019年5月28日下午5:42:34 *类说明:分区 输入数据 Hello HI HI qianfeng hi hi qianfeng qianfeng 163.com qq.com 189.com @163.com @qq.com *123 输出: 将首写字母为A-Z的放到一个文件中 将首写字母为a-z的放到一个文件中 将首写字母为0-9的放到一个文件中 其他的放到一个文件中 结果文件:part-r-00000 Hello 1 Hi 2 结果文件:part-r-00001 hi 2 qianfeng 2 结果文件:part-r-00002 .... 结果文件:part-r-00003 .... */
public class PartitionDemo implements Tool {

	/** * map阶段 * * @author HP * */
	public static class MyMapper extends Mapper<LongWritable, Text, Text, Text> {

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String words[] = line.split(" ");
			for (String s : words) {
				context.write(new Text(s), new Text(1 + ""));
			}
		}

	}

	/** * reduce阶段 */
	public static class MyReducer extends Reducer<Text, Text, Text, Text> {

		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			int counter = 0;
			for (Text t : values) {
				counter += Integer.parseInt(t.toString());
			}
			context.write(key, new Text(counter + ""));
		}

	}

	public void setConf(Configuration conf) {
		// 对conf的属性设置
		conf.set("fs.defaultFS", "hdfs://qf");
		conf.set("dfs.nameservices", "qf");
		conf.set("dfs.ha.namenodes.qf", "nn1, nn2");
		conf.set("dfs.namenode.rpc-address.qf.nn1", "hadoop01:9000");
		conf.set("dfs.namenode.rpc-address.qf.nn2", "hadoop02:9000");
		conf.set("dfs.client.failover.proxy.provider.qf", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
	}

	public Configuration getConf() {
		return new Configuration();
	}

	/** * 驱动方法 */
	public int run(String[] args) throws Exception {
		// 1.获取配置对象信息
		Configuration conf = getConf();
		// 2.对conf进行设置(没有就不用)
		// 3.获取job对象 (注意导入的包)
		Job job = Job.getInstance(conf, "job");
		// 4.设置job的运行主类
		job.setJarByClass(PartitionDemo.class);
		
		//set inputpath and outputpath
		setInputAndOutput(job, conf, args);
		
		
		
		// System.out.println("jiazai finished");
		// 5.对map阶段进行设置
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		//添加分区信息
		job.setPartitionerClass(MyPatitioner.class);
		job.setNumReduceTasks(4);//上面有4种情况 分为4个文件存放
		
		
		// System.out.println("map finished");
		// 6.对reduce阶段进行设置
		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		//提交
		return job.waitForCompletion(true) ? 0 : 1;
		

	}
	
	
	//主方法
	public static void main(String[] args) throws Exception {
		int isok = ToolRunner.run(new Configuration(), new PartitionDemo(), args);
		System.out.println(isok);
	}

	/** * 处理参数的方法 * @param job * @param conf * @param args */
	
	private void setInputAndOutput(Job job, Configuration conf, String[] args) {
		if(args.length != 2) {
			System.out.println("usage:yarn jar /*.jar package.classname /* /*");
			return ;
		}
		//正常处理输入输出参数
		try {
			FileInputFormat.addInputPath(job, new Path(args[0]));
			
			FileSystem fs = FileSystem.get(conf);
			Path outputpath = new Path(args[1]);
			if(fs.exists(outputpath)) {
				fs.delete(outputpath, true);
			}
			FileOutputFormat.setOutputPath(job, outputpath);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}

}

MyPatitioner

package qf.com.mr;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

/* *@author Shishuai E-mail:1198319583@qq.com *@version Create time : 2019年5月28日下午5:39:18 *类说明:自定义的Partitioner *自定义分区需要注意的 *1.分区需要继承Partitioner<key, value>, 其中的key-value需要和map阶段的输出相同 *2.实现getPartition(key, value, numPartitions)方法, 该方法只能返回int类型 *3.分区数和reduce个数相同 *4.默认使用HashPartitioner */
public class MyPatitioner extends Partitioner<Text, Text>{

	@Override
	public int getPartition(Text key, Text value, int numPartitions) {
		String firstChar = key.toString().substring(0,  1);//把第一个字母给截下来
		if(firstChar.matches("^[A-Z]")) {
			return 0%numPartitions;
		}else if(firstChar.matches("^[a-z]")) {
			return 1%numPartitions;
		}else if(firstChar.matches("^[0-9]")) {
			return 2%numPartitions;
		}else {
			return 3%numPartitions;
		}
		
	}
	
}


倒排索引案例
在home下创建三个文件1.html 2.html 3.html
写入数据 数据在代码注释上
然后上传到hdfs
然后(先在hdfs建一个文件夹di 一遍存放这三个文件)

在这里插入图片描述

yarn jar /home/wc.jar qf.com.mr.DescIndexDemo /di /out/20

在这里插入图片描述
在这里插入图片描述
运行完之后查看结果

然后就不对 跟上次分区的输出差不多啊 咋回事
在这里插入图片描述
这个地方忘记改了
卧槽
在这里插入图片描述
还有这个地方
我去
视频上都不说
找了半天

在这里插入图片描述

出现错误
在这里插入图片描述
注意导入的包
解决
在这里插入图片描述


在这里插入图片描述
终于看到了想要的结果


注意:本文归作者所有,未经作者允许,不得转载

全部评论: 0

    我有话说: