hadoop记录

MapReduce的特征
1. 每个分片输入的文件可以比较大。默认64M



Map1结果与Map2结果重叠现象？（传统的分布式计算无法解决）
方案：Map2与Map2原封不动的把数据传到Reduce；　问题：结果Map啥事没干，Reduce最终累死，　分而治之成为了空谈。
最终方案：使用partition把相同key的结果分配到同一个reduce上执行

输入文件把切分成多个块，　每个块的默认大小是64M，　每一个块就是一个maptask任务
map任务工作过程：
Map从ＨＤＦＳ上获取数据块(Datanode上每个map任务都会有一个内存缓冲区（默认１００Ｍ），用来存储它输出的结果「有它的原因：减少对磁盘的读写频率」)
如果map结果很大使缓冲区默认100M满了后将缓冲区中数据spill到磁盘上(spill过程), 默认是80%时就spill(可以设置)，　spill过程中map还产生结果; spill过程
对数据进行sort; spill出来的单独文件（又叫溢写文件）；如果有大数据量的话，就生成多个溢写文件，又要将这些文件合并起来(merge过程「可以通过参数设置每次merge多少个文件」)；　问题：如果merge只是只是合并过程，像WordCount例子,第一个溢写文件中key:name value:4, 第二个溢写文件key:name value:3 的话，merge后还会出现此key值的，所以如果之前设置了combiner，就会使用combiner将相同的key的value进行合并；　最终的这个输出文件也是存储在本地磁盘中。

reduce任务工作过程：
reduce是将map的输出作为reduce的输入，只要有一个map任务执行完就会有reduce任务开始执行。实际运行过程中，map与reduce运行过程很可能不是在同一个datanode上，那它们是通过什么来联系的呢？reduce是通过http向JobTracker发请求看那些map执行完了，然后从map所在的TaskTracker远程获取map的输出文件，注意：并不是获取整个map的输出文件，　而是获取reduce能处理的了的数据; 在这个下载过程中，也会做归并排序，因为不同的map很有可能含有相同的key值; 对于reduce来说它的文件来源可能来自多个map，这个下载过程可以是并行的，不过这个并行数据可以设置的，默认是５，即使reduce 所需要的数据是从很多个map上的（若是大于设置的并行度），也只能一次并行的从设置的这么多个Map上下载所需数据；copy获取到的数据也是放到内存当中，


问题：处理１０亿条数据，只求最大值，　这样的话在网上传输占了很大的带宽／专利计算中最终会把数据聚集到reduce端，加重reduce负担

Combiner：就是把了减少map与reduce间数据传输而设置的；　
注意：combine的输入和reduce的完全一致，输出和map的完全一致；一个combiner只是处理一个结点的输出，不像reduce享受shuffle后的数据;“迷你reduce”

注意几点：
１.与map和reduce不同，combiner没有默认的实现
２.并不是所有的任务都适合使用combiner，比如　求最大值，求和可以，　求中值不行, 平均值也应该不行；
　词频统计map输出(hello, 1), (hello, 1)经过combiner后(hello, 2)压缩后再送到reduce

 

Hadoop将输入数据切分成若干个输入分片(inputsplit)，　并将每个split交给一个MapTask处理，MapTask不断地从对应的split中解析出一个个key/value，　并调用map()函数处理，处理完之后根据ReduceTask个数将结果分成若干个分片写到本地磁盘。同时每个ReduceTask从每个MapTask上读取属于自己的那个partition，　然后使用基于排序的方法将key相同的数据聚焦在一起，调用reduce()函数处理，并将结果输出到文件中。
以上缺少三个基本的组件：１.指定输入文件格式(InputFormat)　　２.map()函数产生的每个key/value对　发给哪个ReduceTask函数处理(Partitioner)　　　３.指定输出文件格式(OutputFormat)

编程实例：
１.Top K问题　　需要统计最近最热门的key个查询词
　　分解成两个MapReduce作业：分别完成统计词频　和　找出词频最高的前K个
　　第一个作业是典型的WordCount问题，
    hello 1
    world 9
    name 3
    tom 5
    john 23 
    对于第二个作业，首先map()函数中输出前key个频率最高的词，然后由reduce()函数汇总每个Map任务得到的前K个查询词，并输出频率最高的前K个查询词
        map输入 9 world 
        reduce收到所有９的单词，１.直接换了kv位置输出 2.通过TreeMap把前k放到集合中，
		public class MyInt implements Comparable<MyInt>{
    		private Integer value;
   		 public MyInt(Integer value){
      		  this.value = value;
   		 }
   		 public int getValue() {
       		 return value;
  		  }
  		  public void setValue(int value) {
  		      this.value = value;
   		 }
  		  @Override
   		 public int compareTo(MyInt o) {
   		     return value.compareTo(o.getValue());
   		 }
		}
　		// 用TreeMap存储可以利用它的排序功能		
		// 这里用 MyInt 因为TreeMap是对key排序，且不能唯一，而词频可能相同，要以词频为Key就必需对它封装
		private static TreeMap<MyInt, String> tm = new TreeMap<MyInt, String>(
				new Comparator<MyInt>()
				{
					/**
					 * 默认是从小到大的顺序排的，现在修改为从大到小
					 * 
					 * @param o1
					 * @param o2
					 * @return
					 */
					@Override
					public int compare(MyInt o1, MyInt o2)
					{
						return o2.compareTo(o1);
					}

				});
		private static MultipleOutputs<Text, IntWritable> mos = null;
		protected void reduce(IntWritable key, Iterable<Text> values,
				Context context) throws IOException, InterruptedException
		{
			for (Text text : values)
			{
				context.write(text, key);
				tm.put(new MyInt(key.get()), text.toString());

				// TreeMap以对内部数据进行了排序，最后一个必定是最小的
				if (tm.size() > k)
				{
					tm.remove(tm.lastKey());
				}

			}
		}
		protected void cleanup(Context context) throws IOException,
				InterruptedException
		{
			String path = context.getConfiguration().get("topKout");
			mos = new MultipleOutputs<Text, IntWritable>(context);
			Set<Entry<MyInt, String>> set = tm.entrySet();
			for (Entry<MyInt, String> entry : set)
			{
				mos.write("topKMOS", new Text(entry.getValue()),
						new IntWritable(entry.getKey().getValue()), path);
			}
			mos.close();
		}