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
运行完之后查看结果
然后就不对 跟上次分区的输出差不多啊 咋回事
这个地方忘记改了
卧槽
还有这个地方
我去
视频上都不说
找了半天
出现错误
注意导入的包
解决
终于看到了想要的结果
注意：本文归作者所有，未经作者允许，不得转载
