MapReduce高级编程2

曾经终败给现在 2022-04-25 02:16 213阅读 0赞

MapReduce Top N 、二次排序，MapJoin：

## TOP N ##

对于一组输入List(key,value),我们要创建一个Top N 列表,这是一种过滤模式,查看输入数据特定子集,观察用户的行为。

### 解决方案 ###

key是唯一键，需要对输入进行额外的聚集处理，先把输入分区成小块，然后把每个小块发送到一个映射器中。每个映射器会创建一个本地Top N 列表发送到一个规约器中，即最终由一个规约其产生一个Top N 列表。对于大多数的MapReduce算法，由一个规约器接收所有数据会使负载不均衡，从而产生瓶颈问题。但是本解决方案产生的Top N 是很少量的数据，如果有映射器，则会需要处理1000 X N的数据。

### 数据 ###

![FWGp9I.png][]

### 思路 ###

计算过程以K V对传输

可以创建一个最小堆，最常用的是SortMap<K,V> 和TreeMap<K,V>，如果top N.size()>N，则删除第一个元素（频数最小）如果求Bottom N构造最大堆。

![FWGL2q.png][]

### 代码 ###

#### Mapper ####

#### Reduce ####

#### Job ####

#### 结果 ####

![FWYanK.png][]

## 非唯一键解决方案 ##

假设所有的输入键是不唯一的，即会出现key相同而value不同的情况，则需要为两个阶段解决。

第一阶段：通过MapReduce先聚集具有相同键的元组，并将频数相加，将不唯一的键转换为唯一的键，其value值为频数和，输出为唯一键值对<key,value>

第二阶段：将第一阶段的输出的唯一键值对作为输入，采用唯一键的解决方案求Top N 即可。

## 全排序 ##

创建随机数据

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>for i in {1..100000}; do echo $RANDOM done; </pre></td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 </pre></td> 
 <td><pre>sh createdata.sh &gt;data1 sh createdata.sh &gt;data2 sh createdata.sh &gt;data3 sh createdata.sh &gt;data4 </pre></td> 
 </tr> 
 </tbody> 
</table>

### MyPartitioner ###

### Mysort ###

### SimpleMapper ###

### SimpleReducer ###

### SimpleJob ###

## 抽样实现全局排序 ##

自定分区下实现用户按业务分区与Reduce任务数据一 一对于,使用全排序最终的解结果在文件中一次按顺序输出,可能存在认为对数据了解不够多，造成分区的数据个数不一致，导致发生数据倾斜的问题。

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 </pre></td> 
 <td><pre>import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.NullWritable; 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.input.KeyValueTextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.partition.InputSampler; import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; public class TotalOrderingPartition extends Configured implements Tool {
 //Map类 static class SimpleMapper extends Mapper&lt;Text, Text, Text, IntWritable&gt; {
 @Override protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
 IntWritable intWritable = new IntWritable(Integer.parseInt(key.toString())); context.write(key, intWritable); } } //Reduce类 static class SimpleReducer extends Reducer&lt;Text, IntWritable, IntWritable, NullWritable&gt; {
 @Override protected void reduce(Text key, Iterable&lt;IntWritable&gt; values, Context context) throws IOException, InterruptedException {
 for (IntWritable value : values) {
 context.write(value, NullWritable.get()); } } } public int run(String[] args) throws Exception {
 Configuration conf = getConf(); Job job = Job.getInstance(conf); job.setJarByClass(TotalOrderingPartition.class); job.setInputFormatClass(KeyValueTextInputFormat.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); job.setNumReduceTasks(3); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(NullWritable.class); TotalOrderPartitioner.setPartitionFile(job.getConfiguration(), new Path(args[2])); InputSampler.Sampler&lt;Text, Text&gt; sampler = new InputSampler.SplitSampler&lt;Text, Text&gt;(1000, 10); InputSampler.writePartitionFile(job, sampler); job.setPartitionerClass(TotalOrderPartitioner.class); job.setMapperClass(SimpleMapper.class); job.setReducerClass(SimpleReducer.class); job.setJobName("Ite Blog"); return job.waitForCompletion(true) ? 0 : 1; } //主方法 public static void main(String[] args) throws Exception {
 args = new String[]{
 "/input/simple","/output/random_simple1","/output/random_simple2"}; int exitCode = ToolRunner.run(new TotalOrderingPartition(), args); System.exit(exitCode); } } </pre></td> 
 </tr> 
 </tbody> 
</table>

通过抽样分区，可以使个分区所含的记录大致相等，使作业的总体执行时间不会因为一个Reducer任务之后而拖慢整体进度。

## 二次排序 ##

二次排序不同于全排序，它在规约阶段对与中间键相关联的中间值的某个属性进行排序。可以对传如各个Reducer的值进行升序或降序排序，即首先按照第一个字段排序，在对第二个字段相同的排序，且不能破坏第一个排序结果。

### 解决方案 ###

1.  Reducer内排序。让Reducer接收（k2,list(v2)）之后,利用Reducer内存进行排序，生成(k2,list(v2)),在进行处理，在数值过多的情况下，容易导致内存溢出。
2.  利用MapReduce框架的排序框架对Reducer值进行排序。为自然键增加自然值得部分或整个值，构造组合键进行排序。由于排序工作是由MapReduce框架完成的，因此不会溢出。

#### 流程 ####

1.  构造组合键(k,v1),其中v1是次键，K为自然键值，要在Reduce中注入一个值（v1）或者一个组合键。Reducer值按照什么来排序，就将其加入到自然键中，共同称为组合键。
    
    对于Key=k2，若所有的映射器生成键值对为(k2,A1),(k2,A2),…(k2,An),对于一每个Ai，对于一个每一个Ai，设Ai为一个m元组值(ai2,ai3,…,aim)用bi来表示。因而映射器生成键值对可以做如下表示:
    
    (k2,(a1,b1)),(k2(a2,b2)),…(k2,(an,bn))
    
    自然键位k2,加入ai形成组合键((k2,ai))最终的映射器发出键值对如下：
    
    ((k2,a1),(a1,b1)),((k2,a2),(a2,b2)),….((k2,an),(an,bn))
2.  在分区器中加入两个插件类：定制分区器，确保相同自然键的数据到达相同Reducer，也就是说，同一Reducer中可能由key=k2,key=k3的所有数据，定制比较器就将这些数据按照自然键进行分组。
3.  分组比较器，可以使v1按有序的顺序到达Reducer，使用MapReduce执行框架完成排序，可以保证到达Reducer的值按键有序并按值有序。

#### 实现类 ####

##### MyWritable #####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 </pre></td> 
 <td><pre>import java.io.*; import org.apache.hadoop.io.*; public class MyWritable implements WritableComparable&lt;MyWritable&gt; {
 private int first; private int second; public MyWritable() {
 } public MyWritable(int first, int second) {
 set(first, second); } public void set(int first, int second) {
 this.first = first; this.second = second; } public int getFirst() {
 return first; } public int getSecond() {
 return second; } public void write(DataOutput out) throws IOException {
 out.writeInt(first); out.writeInt(second); } public void readFields(DataInput in) throws IOException {
 first = in.readInt(); second = in.readInt(); } @Override public int hashCode() {
 return first * 163 + second; } @Override public boolean equals(Object o) {
 if (o instanceof MyWritable) {
 MyWritable ip = (MyWritable) o; return first == ip.first &amp;&amp; second == ip.second; } return false; } @Override public String toString() {
 return first + "\t" + second; } public int compareTo(MyWritable ip) {
 int cmp = compare(first, ip.first); if (cmp != 0) {
 return cmp; } return compare(second, ip.second); } public static int compare(int a, int b) {
 return (a &lt; b ? -1 : (a == b ? 0 : 1)); } } </pre></td> 
 </tr> 
 </tbody> 
</table>

##### MyComparator #####

##### MyPartitioner #####

##### MyGroup #####

##### SecondarySortMapper #####

##### SecondarySortReducer #####

##### SecondarySortJob #####

#### 原始数据 ####

##### 排序后数据 #####

## 连接 ##

### Reduce端连接 ###

#### TableBean ####

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 </pre></td> 
 <td><pre>import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class TableBean implements Writable {
 private String orderId; private String pid; private int amount; private String pname; private String flag; public TableBean() {
 super(); } public TableBean(String orderId, String pid, int amount, String pname, String flag) {
 super(); this.orderId = orderId; this.pid = pid; this.amount = amount; this.pname = pname; this.flag = flag; } @Override public String toString() {
 return orderId + "\t" + pname + "\t" + amount; } public String getOrderId() {
 return orderId; } public void setOrderId(String orderId) {
 this.orderId = orderId; } public String getPid() {
 return pid; } public void setPid(String pid) {
 this.pid = pid; } public int getAmount() {
 return amount; } public void setAmount(int amount) {
 this.amount = amount; } public String getPname() {
 return pname; } public void setPname(String pname) {
 this.pname = pname; } public String getFlag() {
 return flag; } public void setFlag(String flag) {
 this.flag = flag; } @Override public void write(DataOutput out) throws IOException {
 out.writeUTF(orderId); out.writeUTF(pid); out.writeInt(amount); out.writeUTF(pname); out.writeUTF(flag); } @Override public void readFields(DataInput in) throws IOException {
 orderId = in.readUTF(); pid = in.readUTF(); amount = in.readInt(); pname = in.readUTF(); flag = in.readUTF(); } } </pre></td> 
 </tr> 
 </tbody> 
</table>

#### TableMapper ####

#### TableReducer ####

#### TableDriver ####

##### 原始数据 #####

![Ffw9q1.png][]

##### 结果 #####

### Map端连接 ###

##### DistributedCacheMapper #####

##### DistributedCacheDriver #####

这种方法的运行速度很快,输入数据进行Mapper计算前分了部分,启用了一个Mapper,而Reducer连接时启动了两个Mapper.但该种计算模式受到JVM种堆的分配限制,如果集群中的内存足够大,业务符合逻辑(如内连接和做外连接)时可以考虑这种计算模式,如果数据集很大,Reducer连接操作是一个很不错的选择.当然连接操作的思想还有一些其他的方法,可以根据业务不同,采取不同的编程思想。

![Ff0BnA.png][]

[FWGp9I.png]: /images/20220210/fc910a33b7d841ee81b770c044a3925d.png
[FWGL2q.png]: /images/20220210/26a321199a0a4aa4b00b3e4652f95fbd.png
[FWYanK.png]: /images/20220210/5243c9991f764d8ebfe7b987a88979aa.png
[Ffw9q1.png]: /images/20220210/41a00cc82f414038b30ec695a7fa80a6.png
[Ff0BnA.png]: /images/20220210/0dc818aeacbb4d08a99237a9df0fb51f.png