9_大数据之Yarn

一 `Yarn`资源调度器

Yarn是一个资源调度平台，负责为运算程序提供服务器运算资源，相当于一个分布式的操作系统平台，而MapReduce等运算程序则相当于运行于操作系统之上的应用程序。

1️⃣Yarn基本架构 : YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成，如下图所示。

2️⃣Yarn工作机制

（1）MR程序提交到客户端所在的节点。
（2）YarnRunner向ResourceManager申请一个Application。
（3）RM将该应用程序的资源路径返回给YarnRunner。
（4）该程序将运行所需资源提交到HDFS上。
（5）程序资源提交完毕后，申请运行mrAppMaster。
（6）RM将用户的请求初始化成一个Task。
（7）其中一个NodeManager领取到Task任务。
（8）该NodeManager创建容器Container，并产生MRAppmaster。
（9）Container从HDFS上拷贝资源到本地。
（10）MRAppmaster向RM申请运行MapTask资源。
（11）RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（12）MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
（13）MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
（14）ReduceTask向MapTask获取相应分区的数据。
（15）程序运行完毕后，MR会向RM申请注销自己。

3️⃣作业提交全过程

作业提交全过程详解
（1）作业提交
第1步：Client调用job.waitForCompletion方法，向整个集群提交MapReduce作业。
第2步：Client向RM申请一个作业id。
第3步：RM给Client返回该job资源的提交路径和作业id。
第4步：Client提交jar包、切片信息和配置文件到指定的资源提交路径。
第5步：Client提交完资源后，向RM申请运行MrAppMaster。
（2）作业初始化
第6步：当RM收到Client的请求后，将该job添加到容量调度器中。
第7步：某一个空闲的NM领取到该Job。
第8步：该NM创建Container，并产生MRAppmaster。
第9步：下载Client提交的资源到本地。
（3）任务分配
第10步：MrAppMaster向RM申请运行多个MapTask任务资源。
第11步：RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。
（4）任务运行
第12步：MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。
第13步：MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。
第14步：ReduceTask向MapTask获取相应分区的数据。
第15步：程序运行完毕后，MR会向RM申请注销自己。
（5）进度和状态更新
YARN中的任务将其进度和状态(包括counter)返回给应用管理器, 客户端每秒(通过mapreduce.client.progressmonitor.pollinterval设置)向应用管理器请求进度更新, 展示给用户。
（6）作业完成
除了向应用管理器请求作业进度外, 客户端每5秒都会通过调用waitForCompletion()来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval来设置。作业完成之后, 应用管理器和Container会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。

4️⃣作业提交过程之MapReduce

5️⃣资源调度器
目前Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler。Hadoop2.7.2默认的资源调度器是Capacity Scheduler。具体设置详见：yarn-default.xml文件
<property>
   <description>The class to use as the resource scheduler.</description>
   <name>yarn.resourcemanager.scheduler.class</name>
  <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>
1．先进先出调度器（FIFO）

2．容量调度器（Capacity Scheduler）

3．公平调度器（Fair Scheduler）

6️⃣任务的推测执行
1．作业完成时间取决于最慢的任务完成时间
一个作业由若干个Map任务和Reduce任务构成。因硬件老化、软件Bug等，某些任务可能运行非常慢。
思考：系统中有99%的Map任务都完成了，只有少数几个Map老是进度很慢，完不成，怎么办？
2．推测执行机制
发现拖后腿的任务，比如某个任务运行速度远慢于任务平均速度。为拖后腿任务启动一个备份任务，同时运行。谁先运行完，则采用谁的结果。
3．执行推测任务的前提条件
（1）每个Task只能有一个备份任务
（2）当前Job已完成的Task必须不小于0.05（5%）
（3）开启推测执行参数设置。mapred-site.xml文件中默认是打开的。
<property>
  <name>mapreduce.map.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
</property>

<property>
  <name>mapreduce.reduce.speculative</name>
  <value>true</value>
  <description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
</property>
4．不能启用推测执行机制情况
（1）任务间存在严重的负载倾斜；
（2）特殊任务，比如任务向数据库中写数据.

5．算法原理，如下图所示

二 `Hadoop`企业优化

1️⃣MapReduce 跑的慢的原因

2️⃣MapReduce优化方法
MapReduce优化方法主要从六个方面考虑：数据输入、Map阶段、Reduce阶段、IO传输、数据倾斜问题和常用的调优参数。

3️⃣ 常用的调优参数
1．资源相关参数
（1）以下参数是在用户自己的MR应用程序中配置就可以生效（mapred-default.xml）

（2）应该在YARN启动之前就配置在服务器的配置文件中才能生效（yarn-default.xml）

（3）Shuffle性能优化的关键参数，应在YARN启动之前就配置好（mapred-default.xml）

2．容错相关参数(MapReduce性能优化)

4️⃣HDFS小文件优化方法
1 HDFS小文件弊端
HDFS上每个文件都要在NameNode上建立一个索引，这个索引的大小约为150byte，这样当小文件比较多的时候，就会产生很多的索引文件，一方面会大量占用NameNode的内存空间，另一方面就是索引文件过大使得索引速度变慢。
2 HDFS小文件解决方案
（1）在数据采集的时候，就将小文件或小批数据合成大文件再上传HDFS。
（2）在业务处理之前，在HDFS上使用MapReduce程序对小文件进行合并。
（3）在MapReduce处理时，可采用CombineTextInputFormat提高效率

三 `MapReduce`扩展案例

1️⃣倒排索引案例（多job串联）
1．需求 : 有大量的文本（文档、网页），需要建立搜索索引.
（1）数据输入

//a.txt
xxx pingping
xxx ss
xxx ss

//b.txt
xxx pingping
xxx pingping
pingping ss

//c.txt
xxx ss
xxx pingping

（2）期望输出数据

xxx    c.txt-->2   b.txt-->2   a.txt-->3   
pingping   c.txt-->1   b.txt-->3   a.txt-->1   
ss c.txt-->1   b.txt-->1   a.txt-->2

2．需求分析

3．第一次处理
（1）第一次处理，编写OneIndexMapper类

package com.xxx.mapreduce.index;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class OneIndexMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
  
  String name;
  Text k = new Text();
  IntWritable v = new IntWritable();
  
  @Override
  protected void setup(Context context)throws IOException, InterruptedException {

      // 获取文件名称
      FileSplit split = (FileSplit) context.getInputSplit();
      
      name = split.getPath().getName();
  }
  
  @Override
  protected void map(LongWritable key, Text value, Context context)   throws IOException, InterruptedException {

      // 1 获取1行
      String line = value.toString();
      
      // 2 切割
      String[] fields = line.split(" ");
      
      for (String word : fields) {

          // 3 拼接
          k.set(word+"--"+name);
          v.set(1);
          
          // 4 写出
          context.write(k, v);
      }
  }
}

（2）第一次处理，编写OneIndexReducer类

package com.xxx.mapreduce.index;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class OneIndexReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
  
IntWritable v = new IntWritable();

  @Override
  protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
      
      int sum = 0;

      // 1 累加求和
      for(IntWritable value: values){
          sum +=value.get();
      }
      
      v.set(sum);

      // 2 写出
      context.write(key, v);
  }
}

（3）第一次处理，编写OneIndexDriver类

package com.xxx.mapreduce.index;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 OneIndexDriver {

  public static void main(String[] args) throws Exception {

      // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
      args = new String[] { "e:/input/inputoneindex", "e:/output5" };

      Configuration conf = new Configuration();

      Job job = Job.getInstance(conf);
      job.setJarByClass(OneIndexDriver.class);

      job.setMapperClass(OneIndexMapper.class);
      job.setReducerClass(OneIndexReducer.class);

      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(IntWritable.class);
      
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);

      FileInputFormat.setInputPaths(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));

      job.waitForCompletion(true);
  }
}

（4）查看第一次输出结果

xxx--a.txt 3
xxx--b.txt 2
xxx--c.txt 2
pingping--a.txt    1
pingping--b.txt    3
pingping--c.txt    1
ss--a.txt  2
ss--b.txt  1
ss--c.txt  1

4．第二次处理
（1）第二次处理，编写TwoIndexMapper类

package com.xxx.mapreduce.index;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TwoIndexMapper extends Mapper<LongWritable, Text, Text, Text>{

  Text k = new Text();
  Text v = new Text();
  
  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
      
      // 1 获取1行数据
      String line = value.toString();
      
      // 2用“--”切割
      String[] fields = line.split("--");
      
      k.set(fields[0]);
      v.set(fields[1]);
      
      // 3 输出数据
      context.write(k, v);
  }
}

（2）第二次处理，编写TwoIndexReducer类

package com.xxx.mapreduce.index;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class TwoIndexReducer extends Reducer<Text, Text, Text, Text> {

Text v = new Text();

  @Override
  protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
      // xxx a.txt 3
      // xxx b.txt 2
      // xxx c.txt 2

      // xxx c.txt-->2 b.txt-->2 a.txt-->3

      StringBuilder sb = new StringBuilder();

       // 1 拼接
      for (Text value : values) {
          sb.append(value.toString().replace("\t", "-->") + "\t");
      }

v.set(sb.toString());

      // 2 写出
      context.write(key, v);
  }
}

（3）第二次处理，编写TwoIndexDriver类

package com.xxx.mapreduce.index;
import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class TwoIndexDriver {

  public static void main(String[] args) throws Exception {

      // 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "e:/input/inputtwoindex", "e:/output6" };

      Configuration config = new Configuration();
      Job job = Job.getInstance(config);

job.setJarByClass(TwoIndexDriver.class);
      job.setMapperClass(TwoIndexMapper.class);
      job.setReducerClass(TwoIndexReducer.class);

      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(Text.class);
      
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);

      FileInputFormat.setInputPaths(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));

      boolean result = job.waitForCompletion(true);
System.exit(result?0:1);
  }
}

（4）第二次查看最终结果

xxx    c.txt-->2   b.txt-->2   a.txt-->3   
pingping   c.txt-->1   b.txt-->3   a.txt-->1   
ss c.txt-->1   b.txt-->1   a.txt-->2

2️⃣TopN案例

需求 : 对需求2.3输出结果进行加工，输出流量使用量在前10的用户信息;
（1）输入数据

13470253144 180     180     360
13509468723 7335    110349  117684
13560439638 918     4938    5856
13568436656 3597    25635   29232
13590439668 1116    954     2070
13630577991 6960    690     7650
13682846555 1938    2910    4848
13729199489 240     0       240
13736230513 2481    24681   27162
13768778790 120     120     240
13846544121 264     0       264
13956435636 132     1512    1644
13966251146 240     0       240
13975057813 11058   48243   59301
13992314666 3008    3720    6728
15043685818 3659    3538    7197
15910133277 3156    2936    6092
15959002129 1938    180     2118
18271575951 1527    2106    3633
18390173782 9531    2412    11943
84188413    4116    1432    5548

（2）输出数据

13509468723 7335    110349  117684
13975057813 11058   48243   59301
13568436656 3597    25635   29232
13736230513 2481    24681   27162
18390173782 9531    2412    11943
13630577991 6960    690     7650
15043685818 3659    3538    7197
13992314666 3008    3720    6728
15910133277 3156    2936    6092
13560439638 918     4938    5856

需求分析

3．实现代码
（1）编写FlowBean类

package com.xxx.mr.top;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class FlowBean implements WritableComparable<FlowBean>{

   private long upFlow;
   private long downFlow;
   private long sumFlow;
   
   
   public FlowBean() {
       super();
   }

   public FlowBean(long upFlow, long downFlow) {
       super();
       this.upFlow = upFlow;
       this.downFlow = downFlow;
   }

   @Override
   public void write(DataOutput out) throws IOException {
       out.writeLong(upFlow);
       out.writeLong(downFlow);
       out.writeLong(sumFlow);
   }

   @Override
   public void readFields(DataInput in) throws IOException {
       upFlow = in.readLong();
       downFlow = in.readLong();
       sumFlow = in.readLong();
   }

   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;
   }

   @Override
   public String toString() {
       return upFlow + "\t" + downFlow + "\t" + sumFlow;
   }

   public void set(long downFlow2, long upFlow2) {
       downFlow = downFlow2;
       upFlow = upFlow2;
       sumFlow = downFlow2 + upFlow2;
   }

   @Override
   public int compareTo(FlowBean bean) {
       
       int result;
       
       if (this.sumFlow > bean.getSumFlow()) {
           result = -1;
       }else if (this.sumFlow < bean.getSumFlow()) {
           result = 1;
       }else {
           result = 0;
       }
       
       return result;
   }
}

（2）编写TopNMapper类

package com.xxx.mr.top;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopNMapper extends Mapper<LongWritable, Text, FlowBean, Text>{
   
   // 定义一个TreeMap作为存储数据的容器（天然按key排序）
   private TreeMap<FlowBean, Text> flowMap = new TreeMap<FlowBean, Text>();
   private FlowBean kBean;
   
   @Override
   protected void map(LongWritable key, Text value, Context context)   throws IOException, InterruptedException {
       
       kBean = new FlowBean();
       Text v = new Text();
       
       // 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]);
       long sumFlow = Long.parseLong(fields[3]);
       
       kBean.setDownFlow(downFlow);
       kBean.setUpFlow(upFlow);
       kBean.setSumFlow(sumFlow);
       
       v.set(phoneNum);
       
       // 4 向TreeMap中添加数据
       flowMap.put(kBean, v);
       
       // 5 限制TreeMap的数据量，超过10条就删除掉流量最小的一条数据
       if (flowMap.size() > 10) {
//      flowMap.remove(flowMap.firstKey());
           flowMap.remove(flowMap.lastKey());      
}
   }
   
   @Override
   protected void cleanup(Context context) throws IOException, InterruptedException {
       
       // 6 遍历treeMap集合，输出数据
       Iterator<FlowBean> bean = flowMap.keySet().iterator();

       while (bean.hasNext()) {

           FlowBean k = bean.next();

           context.write(k, flowMap.get(k));
       }
   }
}

（3）编写TopNReducer类

package com.xxx.mr.top;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

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

public class TopNReducer extends Reducer<FlowBean, Text, Text, FlowBean> {

   // 定义一个TreeMap作为存储数据的容器（天然按key排序）
   TreeMap<FlowBean, Text> flowMap = new TreeMap<FlowBean, Text>();

   @Override
   protected void reduce(FlowBean key, Iterable<Text> values, Context context)throws IOException, InterruptedException {

       for (Text value : values) {

            FlowBean bean = new FlowBean();
            bean.set(key.getDownFlow(), key.getUpFlow());

            // 1 向treeMap集合中添加数据
           flowMap.put(bean, new Text(value));

           // 2 限制TreeMap数据量，超过10条就删除掉流量最小的一条数据
           if (flowMap.size() > 10) {
               // flowMap.remove(flowMap.firstKey());
flowMap.remove(flowMap.lastKey());
           }
       }
   }

   @Override
   protected void cleanup(Reducer<FlowBean, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {

       // 3 遍历集合，输出数据
       Iterator<FlowBean> it = flowMap.keySet().iterator();

       while (it.hasNext()) {

           FlowBean v = it.next();

           context.write(new Text(flowMap.get(v)), v);
       }
   }
}

（4）编写TopNDriver类

package com.xxx.mr.top;

import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class TopNDriver {

   public static void main(String[] args) throws Exception {
       
       args  = new String[]{"e:/output1","e:/output3"};
       
       // 1 获取配置信息，或者job对象实例
       Configuration configuration = new Configuration();
       Job job = Job.getInstance(configuration);

       // 6 指定本程序的jar包所在的本地路径
       job.setJarByClass(TopNDriver.class);

       // 2 指定本业务job要使用的mapper/Reducer业务类
       job.setMapperClass(TopNMapper.class);
       job.setReducerClass(TopNReducer.class);

       // 3 指定mapper输出数据的kv类型
       job.setMapOutputKeyClass(FlowBean.class);
       job.setMapOutputValueClass(Text.class);

       // 4 指定最终输出的数据的kv类型
       job.setOutputKeyClass(Text.class);
       job.setOutputValueClass(FlowBean.class);

       // 5 指定job的输入原始文件所在目录
       FileInputFormat.setInputPaths(job, new Path(args[0]));
       FileOutputFormat.setOutputPath(job, new Path(args[1]));

       // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
       boolean result = job.waitForCompletion(true);
       System.exit(result ? 0 : 1);
   }
}

3️⃣找博客共同好友案例
1．需求 : 以下是博客的好友列表数据，冒号前是一个用户，冒号后是该用户的所有好友（数据中的好友关系是单向的）求出哪些人两两之间有共同好友，及他俩的共同好友都有谁？
（1）数据输入

A:B,C,D,F,E,O
B:A,C,E,K
C:F,A,D,I
D:A,E,F,L
E:B,C,D,M,L
F:A,B,C,D,E,O,M
G:A,C,D,E,F
H:A,C,D,E,O
I:A,O
J:B,O
K:A,C,D
L:D,E,F
M:E,F,G
O:A,H,I,J

2．需求分析 : 先求出A、B、C、….等是谁的好友,第一次输出结果

A   I,K,C,B,G,F,H,O,D,
B   A,F,J,E,
C   A,E,B,H,F,G,K,
D   G,C,K,A,L,F,E,H,
E   G,M,L,H,A,F,B,D,
F   L,M,D,C,G,A,
G   M,
H   O,
I   O,C,
J   O,
K   B,
L   D,E,
M   E,F,
O   A,H,I,J,F,

第二次输出结果

A-B E C 
A-C D F 
A-D E F 
A-E D B C 
A-F O B C D E 
A-G F E C D 
A-H E C D O 
A-I O 
A-J O B 
A-K D C 
A-L F E D 
A-M E F 
B-C A 
B-D A E 
B-E C 
B-F E A C 
B-G C E A 
B-H A E C 
B-I A 
B-K C A 
B-L E 
B-M E 
B-O A 
C-D A F 
C-E D 
C-F D A 
C-G D F A 
C-H D A 
C-I A 
C-K A D 
C-L D F 
C-M F 
C-O I A 
D-E L 
D-F A E 
D-G E A F 
D-H A E 
D-I A 
D-K A 
D-L E F 
D-M F E 
D-O A 
E-F D M C B 
E-G C D 
E-H C D 
E-J B 
E-K C D 
E-L D 
F-G D C A E 
F-H A D O E C 
F-I O A 
F-J B O 
F-K D C A 
F-L E D 
F-M E 
F-O A 
G-H D C E A 
G-I A 
G-K D A C 
G-L D F E 
G-M E F 
G-O A 
H-I O A 
H-J O 
H-K A C D 
H-L D E 
H-M E 
H-O A 
I-J O 
I-K A 
I-O A 
K-L D 
K-O A 
L-M E F

3．代码实现
（1）第一次Mapper类

package com.xxx.mapreduce.friends;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class OneShareFriendsMapper extends Mapper<LongWritable, Text, Text, Text>{
   
   @Override
   protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
           throws IOException, InterruptedException {

       // 1 获取一行 A:B,C,D,F,E,O
       String line = value.toString();
       
       // 2 切割
       String[] fields = line.split(":");
       
       // 3 获取person和好友
       String person = fields[0];
       String[] friends = fields[1].split(",");
       
       // 4写出去
       for(String friend: friends){

           // 输出 <好友，人>
           context.write(new Text(friend), new Text(person));
       }
   }
}

（2）第一次Reducer类

package com.xxx.mapreduce.friends;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class OneShareFriendsReducer extends Reducer<Text, Text, Text, Text>{
   
   @Override
   protected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {
       
       StringBuffer sb = new StringBuffer();

       //1 拼接
       for(Text person: values){
           sb.append(person).append(",");
       }
       
       //2 写出
       context.write(key, new Text(sb.toString()));
   }
}

（3）第一次Driver类

package com.xxx.mapreduce.friends;
import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class OneShareFriendsDriver {

   public static void main(String[] args) throws Exception {
       
// 1 获取job对象
       Configuration configuration = new Configuration();
       Job job = Job.getInstance(configuration);
       
       // 2 指定jar包运行的路径
       job.setJarByClass(OneShareFriendsDriver.class);

       // 3 指定map/reduce使用的类
       job.setMapperClass(OneShareFriendsMapper.class);
       job.setReducerClass(OneShareFriendsReducer.class);
       
       // 4 指定map输出的数据类型
       job.setMapOutputKeyClass(Text.class);
       job.setMapOutputValueClass(Text.class);
       
       // 5 指定最终输出的数据类型
       job.setOutputKeyClass(Text.class);
       job.setOutputValueClass(Text.class);
       
       // 6 指定job的输入原始所在目录
       FileInputFormat.setInputPaths(job, new Path(args[0]));
       FileOutputFormat.setOutputPath(job, new Path(args[1]));
       
       // 7 提交
       boolean result = job.waitForCompletion(true);
       
       System.exit(result?0:1);
   }
}

（4）第二次Mapper类

package com.xxx.mapreduce.friends;
import java.io.IOException;
import java.util.Arrays;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TwoShareFriendsMapper extends Mapper<LongWritable, Text, Text, Text>{
   
   @Override
   protected void map(LongWritable key, Text value, Context context)
           throws IOException, InterruptedException {

       // A I,K,C,B,G,F,H,O,D,
       // 友 人，人，人
       String line = value.toString();
       String[] friend_persons = line.split("\t");

       String friend = friend_persons[0];
       String[] persons = friend_persons[1].split(",");

       Arrays.sort(persons);

       for (int i = 0; i < persons.length - 1; i++) {
           
           for (int j = i + 1; j < persons.length; j++) {
               // 发出 <人-人，好友> ，这样，相同的“人-人”对的所有好友就会到同1个reduce中去
               context.write(new Text(persons[i] + "-" + persons[j]), new Text(friend));
           }
       }
   }
}

（5）第二次Reducer类

package com.xxx.mapreduce.friends;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TwoShareFriendsReducer extends Reducer<Text, Text, Text, Text>{
   
   @Override
   protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
       
       StringBuffer sb = new StringBuffer();

       for (Text friend : values) {
           sb.append(friend).append(" ");
       }
       
       context.write(key, new Text(sb.toString()));
   }
}

（6）第二次Driver类

package com.xxx.mapreduce.friends;
import org.apache.hadoop.conf.Configuration;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class TwoShareFriendsDriver {

   public static void main(String[] args) throws Exception {
       
// 1 获取job对象
       Configuration configuration = new Configuration();
       Job job = Job.getInstance(configuration);
       
       // 2 指定jar包运行的路径
       job.setJarByClass(TwoShareFriendsDriver.class);

       // 3 指定map/reduce使用的类
       job.setMapperClass(TwoShareFriendsMapper.class);
       job.setReducerClass(TwoShareFriendsReducer.class);
       
       // 4 指定map输出的数据类型
       job.setMapOutputKeyClass(Text.class);
       job.setMapOutputValueClass(Text.class);
       
       // 5 指定最终输出的数据类型
       job.setOutputKeyClass(Text.class);
       job.setOutputValueClass(Text.class);
       
       // 6 指定job的输入原始所在目录
       FileInputFormat.setInputPaths(job, new Path(args[0]));
       FileOutputFormat.setOutputPath(job, new Path(args[1]));
       
       // 7 提交
       boolean result = job.waitForCompletion(true);
       System.exit(result?0:1);
   }
}

四常见错误及解决方案

1）导包容易出错。尤其Text和CombineTextInputFormat。
2）Mapper中第一个输入的参数必须是LongWritable或者NullWritable，不可以是IntWritable. 报的错误是类型转换异常。
3）java.lang.Exception: java.io.IOException: Illegal partition for 13926435656 (4)，说明Partition和ReduceTask个数没对上，调整ReduceTask个数。
4）如果分区数不是1，但是reducetask为1，不执行分区过程。因为在MapTask的源码中，执行分区的前提是先判断ReduceNum个数是否大于1。不大于1肯定不执行。
5）在Windows环境编译的jar包导入到Linux环境中运行，hadoop jar wc.jar com.xxx.mapreduce.wordcount.WordCountDriver /user/xxx/ /user/xxx/output,报如下错误：Exception in thread "main" java.lang.UnsupportedClassVersionError: com/xxx/mapreduce/wordcount/WordCountDriver : Unsupported major.minor version 52.0原因是Windows环境用的jdk1.7，Linux环境用的jdk1.8。解决方案：统一jdk版本。
6）缓存pd.txt小文件案例中，报找不到pd.txt文件原因：大部分为路径书写错误。还有就是要检查pd.txt.txt的问题。还有个别电脑写相对路径找不到pd.txt，可以修改为绝对路径。
7）报类型转换异常。通常都是在驱动函数中设置Map输出和最终输出时编写错误。Map输出的key如果没有排序，也会报类型转换异常。
8）集群中运行wc.jar时出现了无法获得输入文件。
原因：WordCount案例的输入文件不能放用HDFS集群的根目录。
9）出现了如下相关异常Exception in thread "main" java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)at org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Native Method) at org.apache.hadoop.io.nativeio.NativeIO$Windows.access(NativeIO.java:609) at org.apache.hadoop.fs.FileUtil.canRead(FileUtil.java:977) java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries. at org.apache.hadoop.util.Shell.getQualifiedBinPath(Shell.java:356) at org.apache.hadoop.util.Shell.getWinUtilsPath(Shell.java:371) at org.apache.hadoop.util.Shell.<clinit>(Shell.java:364)
解决方案一：拷贝hadoop.dll文件到Windows目录C:\Windows\System32。个别电脑还需要修改Hadoop源码。
解决方案二：创建如下包名，并将NativeIO.java拷贝到该包名下

10）自定义Outputformat时，注意在RecordWirter中的close方法必须关闭流资源。否则输出的文件内容中数据为空。
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
       if (xxxfos != null) {
           xxxfos.close();
       }
       if (otherfos != null) {
           otherfos.close();
       }
}

9_大数据之Yarn

一 Yarn资源调度器

二 Hadoop企业优化

三 MapReduce扩展案例

四 常见错误及解决方案

一 `Yarn`资源调度器

二 `Hadoop`企业优化

三 `MapReduce`扩展案例

四常见错误及解决方案