广州高档网站建设,网站设计框架,友情链接例子,轻松筹网站可以做吗3.3.1Shuffle机制
Map方法之后#xff0c;Reduce方法之前的数据处理过程称之为Shuffle。 3.3.2Partition分区
1、问题引出
要求将统计结果按照条件输出到不同文件中#xff08;分区#xff09;。比如#xff1a;将统计结果按照手机归属地不同省份输出到不同文件中#…3.3.1Shuffle机制
Map方法之后Reduce方法之前的数据处理过程称之为Shuffle。 3.3.2Partition分区
1、问题引出
要求将统计结果按照条件输出到不同文件中分区。比如将统计结果按照手机归属地不同省份输出到不同文件中分区
2、默认Partitioner分区
public int getPartition(K2 key, V2 value,int numReduceTasks) {return (key.hashCode() Integer.MAX_VALUE) % numReduceTasks;默认分区是根据key的hashCode对ReduceTasks个数取模得到的。用户没法控制哪个key存储到哪个分区。
3、自定义Partitioner步骤
1自定义类继承Partitioner重写getPartition()方法
public class ProvincePartitioner extends PartitionerText, FlowBean {Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {//控制分区代码逻辑...return partition;}
}2在Job驱动中设置自定义Partitioner
job.setPartitionerClass(ProvincePartitioner.class);3自定义Parttition后要根据自定义Partitioner的逻辑设置相应数量的ReduceTask
// 9 同时指定相应数量的reduce task
job.setNumReduceTasks(5);4、分区总结
1如果ReduceTask的数量getPartition的结果数则会多昌盛几个空的输出文件part-r000xx
2如果1ReduceTask的数量getPartition的结果数则有一部分分区数据无处安放会Exception
3如果ReduceTask的数量1则不管MapTask端输出多少个分区文件最终结果都交给这一个ReduceTask最终也就只会产生一个结果文件part-r-00000
4分区号必须从零开始逐一累加。
5、案例分析
例如假设自定义分区数为5则
1job.setNumReduceTasks(1);会正常运行只产生一个输出文件
2job.setNumReduceTasks(2); 会报错
3job.setNumReduceTasks(6); 大于5程序会正常运行会产生空文件
3.3.3Partition分区案例实操
1、需求
将统计结果按照手机归属地不同省份输出到不同文件中分区
1输入数据
1 13736230513 192.196.100.1 www.atguigu.com 2481 24681 200
2 13846544121 192.196.100.2 264 0 200
3 13956435636 192.196.100.3 132 1512 200
4 13966251146 192.168.100.1 240 0 404
5 18271575951 192.168.100.2 www.atguigu.com 1527 2106 200
6 84188413 192.168.100.3 www.atguigu.com 4116 1432 200
7 13590439668 192.168.100.4 1116 954 200
8 15910133277 192.168.100.5 www.hao123.com 3156 2936 200
9 13729199489 192.168.100.6 240 0 200
10 13630577991 192.168.100.7 www.shouhu.com 6960 690 200
11 15043685818 192.168.100.8 www.baidu.com 3659 3538 200
12 15959002129 192.168.100.9 www.atguigu.com 1938 180 500
13 13560439638 192.168.100.10 918 4938 200
14 13470253144 192.168.100.11 180 180 200
15 13682846555 192.168.100.12 www.qq.com 1938 2910 200
16 13992314666 192.168.100.13 www.gaga.com 3008 3720 200
17 13509468723 192.168.100.14 www.qinghua.com 7335 110349 404
18 18390173782 192.168.100.15 www.sogou.com 9531 2412 200
19 13975057813 192.168.100.16 www.baidu.com 11058 48243 200
20 13768778790 192.168.100.17 120 120 200
21 13568436656 192.168.100.18 www.alibaba.com 2481 24681 200
22 13568436656 192.168.100.19 1116 954 2002期望输出数据
手机号136、137、138、139开头都分别放到一个独立的4个文件中其他开头的放到一个文件中。
2、需求分析 3、在案例2.4的基础上增加一个分区类
package com.cuiyf41.flowsum;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class ProvincePartitioner extends PartitionerText, FlowBean {Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {// 1 获取电话号码的前三位String preNum key.toString().substring(0, 3);int partition 4;// 2 判断是哪个省if (136.equals(preNum)){partition 0;}else if (137.equals(preNum)){partition 1;}else if (138.equals(preNum)){partition 2;}else if (139.equals(preNum)){partition 3;}return partition;}
}4、在驱动函数中增加自定义数据分区设置和ReduceTask设置
// 8 指定自定义数据分区
job.setPartitionerClass(ProvincePartitioner.class);// 9 同时指定相应数量的reduce task
job.setNumReduceTasks(5);package com.cuiyf41.flowsum;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.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class FlowsumDriver {public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {// 输入输出路径需要根据自己电脑上实际的输入输出路径设置args new String[] { e:/input/phone_data.txt, e:/output };// 1 获取配置信息或者job对象实例Configuration conf new Configuration();Job job Job.getInstance(conf);// 6 指定本程序的jar包所在的本地路径job.setJarByClass(FlowsumDriver.class);// 2 指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);// 3 指定mapper输出数据的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);// 4 指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);// 5 指定job的输入原始文件所在目录Path input new Path(args[0]);Path output new Path(args[1]);// 如果输出路径存在则进行删除FileSystem fs FileSystem.get(conf);if (fs.exists(output)) {fs.delete(output,true);}FileInputFormat.setInputPaths(job, input);FileOutputFormat.setOutputPath(job, output);// 8 指定自定义数据分区job.setPartitionerClass(ProvincePartitioner.class);// 9 同时指定相应数量的reduce taskjob.setNumReduceTasks(5);// 7 将job中配置的相关参数以及job所用的java类所在的jar包 提交给yarn去运行boolean result job.waitForCompletion(true);System.exit(result ? 0 : 1);}
}3.3.4WritableComparable排序
1、概述
排序是MapReduce框架中最重要的操作之一。
MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序而不管逻辑上是否需要。
默认排序是按照字典顺序排序且实现该排序的方法是快速排序。
对于MapTask它会将处理的结果暂时放到环形缓冲区中当环形缓冲区使用率达到一定阈值时再对缓冲区中的数据进行一次快速排序并将这些有序数据溢写到磁盘上而当数据处理完毕后它会对磁盘上所有文件进行归并排序。
对于ReduceTask它从每个MapTask上远程拷贝相应的数据文件如果文件大小超过一定的阈值则溢写到磁盘上否则存储在内存中。如果磁盘上文件数目达到一定阈值则进行一次归并排序以生成一个更大的文件如果内存中文件大小或者数量超过一定阈值则进行一次合并后将数据溢写到磁盘上。当所有数据拷贝完毕后ReduceTask统一对内存和磁盘上的所有数据进行一次归并排序。
2、排序的分类
1部分排序
MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。
2全排序
最终输出结果只有一个文件且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低因为一台机器处理所有文件完全丧失了MapReduce所提供的并行架构。
3辅助排序GroupingComparator分组
在Reduce端对key进行分组。应用于在接收的key为bean对象时想让一个或几个字段相同全部比较不相同的key进入到同一个reduce方法时可以采用分组排序。
4二次排序
在自定义排序过程中如果compareTo中的判断条件为两个即为二次排序。
3、自定义排序WritableComparable
1原理分析
bean对象做为key传输需要实现WritableComparable接口重写compareTo方法就可以实现排序。
Override
public int compareTo(FlowBean o) {int result;// 按照总流量大小倒序排列if (sumFlow bean.getSumFlow()) {result -1;}else if (sumFlow bean.getSumFlow()) {result 1;}else {result 0;}return result;
}3.3.5WritableComparable排序案例实操全排序
1、需求
根据案例2.3产生的结果再次对总流量进行排序。
1输入数据
原始数据 第一次处理后的数据
2期望输出数据
13509468723 7335 110349 117684
13736230513 2481 24681 27162
13956435636 132 1512 1644
13846544121 264 0 264
。。。 。。。2、需求分析 3、代码实现
1FlowBean对象在在需求1基础上增加了比较功能
package com.cuiyf41.sort;import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class FlowBean implements WritableComparableFlowBean {private long upFlow;private long downFlow;private long sumFlow;// 反序列化时需要反射调用空参构造函数所以必须有public FlowBean() {super();}public FlowBean(long upFlow, long downFlow) {this.upFlow upFlow;this.downFlow downFlow;this.sumFlow upFlow downFlow;}/*** 序列化方法* param out* throws IOException*/Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}/*** 反序列化方法 注意反序列化的顺序和序列化的顺序完全一致* param in* throws IOException*/Overridepublic void readFields(DataInput in) throws IOException {upFlow in.readLong();downFlow in.readLong();sumFlow in.readLong();}Overridepublic String toString() {return upFlow \t downFlow \t sumFlow;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow upFlow;}public long getDownFlow() {return downFlow;}public void setDownFlow(long downFlow) {this.downFlow downFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow sumFlow;}Overridepublic int compareTo(FlowBean o) {int result;// 按照总流量大小倒序排列if (this.sumFlow o.getSumFlow()) {result -1;}else if (this.sumFlow o.getSumFlow()) {result 1;}else {result 0;}return result;}public void set(long upFlow, long downFlow) {this.upFlow upFlow;this.downFlow downFlow;this.sumFlow upFlow downFlow;}
}2编写Mapper类
package com.cuiyf41.sort;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class FlowCountSortMapper extends MapperLongWritable, Text, FlowBean, Text {FlowBean k new FlowBean();Text v new Text();Overrideprotected void map(LongWritable key, Text value, MapperLongWritable, Text, FlowBean, Text.Context context) throws IOException, InterruptedException {// 1 获取一行String line value.toString();// 2 截取String[] fields line.split(\t);// 3 封装对象String phoneNum fields[0];long upFlow Long.parseLong(fields[1]);long downFlow Long.parseLong(fields[2]);k.set(upFlow, downFlow);v.set(phoneNum);// 4 输出context.write(k, v);}
}3编写Reducer类
package com.cuiyf41.sort;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class FlowCountSortReducer extends ReducerFlowBean, Text, Text, FlowBean {Overrideprotected void reduce(FlowBean key, IterableText values, ReducerFlowBean, Text, Text, FlowBean.Context context) throws IOException, InterruptedException {// 循环输出避免总流量相同情况for(Text value: values){context.write(value, key);}}
}4编写Driver类
package com.cuiyf41.flowsum;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class FlowCountMapper extends MapperLongWritable, Text, Text, FlowBean {FlowBean v new FlowBean();Text k new Text();Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 1 获取一行String line value.toString();// 2 切割字段String[] fields line.split(\t);// 3 封装对象// 取出手机号码String phoneNum fields[1];// 取出上行流量和下行流量long upFlow Long.parseLong(fields[fields.length - 3]);long downFlow Long.parseLong(fields[fields.length - 2]);k.set(phoneNum);
// FlowBean v new FlowBean(upFlow, downFlow);v.set(upFlow, downFlow);// 4 写出context.write(k, v);}
}3.3.6WritableComparable排序案例实操区内排序
1需求
要求每个省份手机号输出的文件中按照总流量内部排序。
2需求分析
基于前一个需求增加自定义分区类分区按照省份手机号设置。 3案例实操
1增加自定义分区类
package com.cuiyf41.sort;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class ProvincePartitioner extends PartitionerFlowBean, Text {Overridepublic int getPartition(FlowBean key, Text value, int numPartitions) {// 1 获取手机号码前三位String preNum value.toString().substring(0, 3);int partition 4;// 2 根据手机号归属地设置分区if (136.equals(preNum)) {partition 0;}else if (137.equals(preNum)) {partition 1;}else if (138.equals(preNum)) {partition 2;}else if (139.equals(preNum)) {partition 3;}return partition;}
}2在驱动类中添加分区类
// 加载自定义分区类
job.setPartitionerClass(ProvincePartitioner.class);// 设置Reducetask个数
job.setNumReduceTasks(5);3.3.7Combiner合并
1概述
1Combiner是MR程序中Mapper和Reducer之外的一个组件。
2Combiner组件的父类就是Reducer。
3Combiner和Reducer的区别在于运行的维值
Combiner是在每个MapTask所在的节点运行
Reducer是接收全局所有Mapper的输出结果
4Combiner的意义就是对每一个MapTask的输出进行局部汇总以减少网络传输量。
5Combiner能够应用的前提是不能影响最终的业务逻辑而且Combiner的输出kv应该跟Reducer的输入kv类型保持一致。
Mapper Reducer
3 5 7 →(357)/3 5 (35726)/5 23/5 等于(54)/2 9/2
2 6 →(26)/2 4
2自定义Combiner实现步骤
a自定义一个Combiner继承Reducer重写Reduce方法
public class WordcountCombiner extends ReducerText, IntWritable, Text,IntWritable{Overrideprotected void reduce(Text key, IterableIntWritable values,Context context) throws IOException, InterruptedException {// 1 汇总操作int count 0;for(IntWritable v :values){count v.get();}// 2 写出context.write(key, new IntWritable(count));}
}b在Job驱动类中设置
job.setCombinerClass(WordcountCombiner.class);3.3.8Combiner合并案例实操
1、需求
统计过程中对每一个MapTask的输出进行局部汇总以减小网络传输量即采用Combiner功能。
1数据输入
2期望输出数据
期望Combine输入数据多输出时经过合并输出数据降低。
2、需求分析 3案例实操-方案一
1增加一个WordcountCombiner类继承Reducer
package com.cuiyf41.wordcount;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class WordcountCombiner extends ReducerText, IntWritable, Text, IntWritable {IntWritable v new IntWritable();Overrideprotected void reduce(Text key, IterableIntWritable values, ReducerText, IntWritable, Text, IntWritable.Context context) throws IOException, InterruptedException {// 1 汇总int sum 0;for(IntWritable value :values){sum value.get();}v.set(sum);// 2 写出context.write(key, v);}
}2在WordcountDriver驱动类中指定Combiner
// 指定需要使用combiner以及用哪个类作为combiner的逻辑
job.setCombinerClass(WordcountCombiner.class);4案例实操-方案二
1将WordcountReducer作为Combiner在WordcountDriver驱动类中指定
// 指定需要使用Combiner以及用哪个类作为Combiner的逻辑
job.setCombinerClass(WordcountReducer.class);3.3.9GroupingComparator分组辅助排序
1概述
对Reduce阶段的数据根据某一个或几个字段进行分组。
分组排序步骤
1自定义类继承WritableComparator
2重写compare()方法
Override
public int compare(WritableComparable a, WritableComparable b) {// 比较的业务逻辑return result;
}3创建一个构造将比较对象的类传给父类
protected OrderGroupingComparator() {super(OrderBean.class, true);
}3.3.10GroupingComparator分组案例实操
1、需求
有如下订单数据 现在需要求出每一个订单中最贵的商品。
1输入数据
0000001 Pdt_01 222.8
0000002 Pdt_05 722.4
0000001 Pdt_02 33.8
0000003 Pdt_06 232.8
0000003 Pdt_02 33.8
0000002 Pdt_03 522.8
0000002 Pdt_04 122.42期望输出数据
1 222.8
2 722.4
3 232.82、需求分析
1利用“订单id和成交金额”作为key可以将Map阶段读取到的所有订单数据按照id升序排序如果id相同再按照金额降序排序发送到Reduce。
2在Reduce端利用groupingComparator将订单id相同的kv聚合成组然后取第一个即是该订单中最贵商品如图4-18所示。 3、代码实现
1定义订单信息OrderBean类
package com.cuiyf41.order;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class OrderBean implements WritableComparableOrderBean {private int order_id; // 订单id号private double price; // 价格public OrderBean() {super();}public OrderBean(int order_id, double price) {super();this.order_id order_id;this.price price;}// 二次排序Overridepublic int compareTo(OrderBean o) {int result;if (order_id o.getOrder_id()) {result 1;} else if (order_id o.getOrder_id()) {result -1;} else {// 价格倒序排序result price o.getPrice() ? -1 : 1;}return result;}Overridepublic void write(DataOutput out) throws IOException {out.writeInt(order_id);out.writeDouble(price);}Overridepublic void readFields(DataInput in) throws IOException {order_id in.readInt();price in.readDouble();}Overridepublic String toString() {return order_id \t price;}public int getOrder_id() {return order_id;}public void setOrder_id(int order_id) {this.order_id order_id;}public double getPrice() {return price;}public void setPrice(double price) {this.price price;}
}2编写OrderSortMapper类
package com.cuiyf41.order;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class OrderMapper extends MapperLongWritable, Text, OrderBean, NullWritable {OrderBean k new OrderBean();Overrideprotected void map(LongWritable key, Text value, MapperLongWritable, Text, OrderBean, NullWritable.Context context) throws IOException, InterruptedException {// 1 获取一行String line value.toString();// 2 截取String[] fields line.split(\t);// 3 封装对象k.setOrder_id(Integer.parseInt(fields[0]));k.setPrice(Double.parseDouble(fields[2]));// 4 写出context.write(k, NullWritable.get());}
}3编写OrderSortGroupingComparator类
package com.cuiyf41.order;import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;public class OrderGroupingComparator extends WritableComparator {protected OrderGroupingComparator() {super(OrderBean.class, true);}Overridepublic int compare(WritableComparable a, WritableComparable b) {OrderBean aBean (OrderBean) a;OrderBean bBean (OrderBean) b;int result;if (aBean.getOrder_id() bBean.getOrder_id()) {result 1;} else if (aBean.getOrder_id() bBean.getOrder_id()) {result -1;} else {result 0;}return result;}
}4编写OrderSortReducer类
package com.cuiyf41.order;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class OrderReducer extends ReducerOrderBean, NullWritable, OrderBean, NullWritable {Overrideprotected void reduce(OrderBean key, IterableNullWritable values, ReducerOrderBean, NullWritable, OrderBean, NullWritable.Context context) throws IOException, InterruptedException {context.write(key, NullWritable.get());}
}5编写OrderSortDriver类
package com.cuiyf41.order;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.IOException;public class OrderDriver {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {// 输入输出路径需要根据自己电脑上实际的输入输出路径设置args new String[]{e:/input/inputorder , e:/output1};// 1 获取配置信息Configuration conf new Configuration();Job job Job.getInstance(conf);// 2 设置jar包加载路径job.setJarByClass(OrderDriver.class);// 3 加载map/reduce类job.setMapperClass(OrderMapper.class);job.setReducerClass(OrderReducer.class);// 4 设置map输出数据key和value类型job.setMapOutputKeyClass(OrderBean.class);job.setMapOutputValueClass(NullWritable.class);// 5 设置最终输出数据的key和value类型job.setOutputKeyClass(OrderBean.class);job.setOutputValueClass(NullWritable.class);// 6 设置输入数据和输出数据路径Path input new Path(args[0]);Path output new Path(args[1]);// 如果输出路径存在则进行删除FileSystem fs FileSystem.get(conf);if (fs.exists(output)) {fs.delete(output,true);}FileInputFormat.setInputPaths(job, input);FileOutputFormat.setOutputPath(job, output);// 8 设置reduce端的分组job.setGroupingComparatorClass(OrderGroupingComparator.class);// 7 提交boolean result job.waitForCompletion(true);System.exit(result ? 0 : 1);}
}
