1.准确理解Mapreduce排序的实验原理

2.熟练掌握Mapreduce排序的程序代码编写

3.培养编写MapReduce排序代码解决问题的能力
实验原理

Map、Reduce任务中Shuffle和排序的过程图如下:

流程分析:

1.Map端:

(1)每个输入分片会让一个map任务来处理,默认情况下,以HDFS的一个块的大小(默认为64M)为一个分片,当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中(该缓冲区的大小默认为100M,由io.sort.mb属性控制),当该缓冲区快要溢出时(默认为缓冲区大小的80%,由io.sort.spill.percent属性控制),会在本地文件系统中创建一个溢出文件,将该缓冲区中的数据写入这个文件。

(2)在写入磁盘之前,线程首先根据reduce任务的数目将数据划分为相同数目的分区,也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据,而有些reduce任务却分到很少数据,甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序,如果此时设置了Combiner,将排序后的结果进行Combia操作,这样做的目的是让尽可能少的数据写入到磁盘。

(3)当map任务输出最后一个记录时,可能会有很多的溢出文件,这时需要将这些文件合并。合并的过程中会不断地进行排序和combia操作,目的有两个:①尽量减少每次写入磁盘的数据量。②尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量,这里可以将数据压缩,只要将mapred.compress.map.out设置为true就可以了。

(4)将分区中的数据拷贝给相对应的reduce任务。有人可能会问:分区中的数据怎么知道它对应的reduce是哪个呢?其实map任务一直和其父TaskTracker保持联系,而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。

到这里,map端就分析完了。那到底什么是Shuffle呢?Shuffle的中文意思是“洗牌”,如果我们这样看:一个map产生的数据,结果通过hash过程分区却分配给了不同的reduce任务,是不是一个对数据洗牌的过程呢?

2.Reduce端:

(1)Reduce会接收到不同map任务传来的数据,并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小,则直接存储在内存中(缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制,表示用作此用途的堆空间的百分比),如果数据量超过了该缓冲区大小的一定比例(由mapred.job.shuffle.merge.percent决定),则对数据合并后溢写到磁盘中。

(2)随着溢写文件的增多,后台线程会将它们合并成一个更大的有序的文件,这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端,MapReduce都是反复地执行排序,合并操作,现在终于明白了有些人为什么会说:排序是hadoop的灵魂。

(3)合并的过程中会产生许多的中间文件(写入磁盘了),但MapReduce会让写入磁盘的数据尽可能地少,并且最后一次合并的结果并没有写入磁盘,而是直接输入到reduce函数。

熟悉MapReduce的人都知道:排序是MapReduce的天然特性!在数据达到reducer之前,MapReduce框架已经对这些数据按键排序了。但是在使用之前,首先需要了解它的默认排序规则。它是按照key值进行排序的,如果key为封装的int为IntWritable类型,那么MapReduce按照数字大小对key排序,如果Key为封装String的Text类型,那么MapReduce将按照数据字典顺序对字符排序。

了解了这个细节,我们就知道应该使用封装int的Intwritable型数据结构了,也就是在map这里,将读入的数据中要排序的字段转化为Intwritable型,然后作为key值输出(不排序的字段作为value)。reduce阶段拿到<key,value-list>之后,将输入的key作为的输出key,并根据value-list中的元素的个数决定输出的次数。
实验环境

Linux Ubuntu 14.04

jdk-7u75-linux-x64

hadoop-2.6.0-cdh5.4.5

hadoop-2.6.0-eclipse-cdh5.4.5.jar

eclipse-java-juno-SR2-linux-gtk-x86_64
实验内容

在电商网站上,当我们进入某电商页面里浏览商品时,就会产生用户对商品访问情况的数据 ,名为goods_visit1,goods_visit1中包含(商品id ,点击次数)两个字段,内容以“\t”分割,由于数据量很大,所以为了方便统计我们只截取它的一部分数据,内容如下:
view plain "copy

商品id  点击次数
1010037 100
1010102 100
1010152 97
1010178 96
1010280 104
1010320 103
1010510 104
1010603 96
1010637 97

要求我们编写mapreduce程序来对商品点击次数有低到高进行排序。

实验结果数据如下:
view plain "copy

点击次数 商品ID
96  1010603
96  1010178
97  1010637
97  1010152
100 1010102
100 1010037
103 1010320
104 1010510
104 1010280

实验步骤

1.切换到/apps/hadoop/sbin目录下,开启Hadoop。
view plain "copy

cd /apps/hadoop/sbin
./start-all.sh

2.在Linux本地新建/data/mapreduce3目录。
view plain "copy

mkdir -p /data/mapreduce3

3.在Linux中切换到/data/mapreduce3目录下,用wget命令从http://10.51.46.104:60000/allfiles/mapreduce3/goods_visit1网址上下载文本文件goods_visit1。
view plain "copy

cd /data/mapreduce3
wget http://10.51.46.104:60000/allfiles/mapreduce3/goods_visit1

然后在当前目录下用wget命令从http://10.51.46.104:60000/allfiles/mapreduce3/hadoop2lib.tar.gz网址上下载项目用到的依赖包。
view plain "copy

wget http://10.51.46.104:60000/allfiles/mapreduce3/hadoop2lib.tar.gz

将hadoop2lib.tar.gz解压到当前目录下。
view plain "copy

tar zxvf hadoop2lib.tar.gz

4.首先在HDFS上新建/mymapreduce3/in目录,然后将Linux本地/data/mapreduce3目录下的goods_visit1文件导入到HDFS的/mymapreduce3/in目录中。
view plain "copy

hadoop fs -mkdir -p /mymapreduce3/in
hadoop fs -put /data/mapreduce3/goods_visit1 /mymapreduce3/in

5.新建Java Project项目,项目名为mapreduce3。

在mapreduce3项目下新建包,包名为mapreduce。

在mapreduce包下新建类,类名为OneSort。

6.添加项目所需依赖的jar包,右键单击项目新建一个文件夹,名为hadoop2lib,用于存放项目所需的jar包。

将/data/mapreduce3目录下hadoop2lib文件夹中的所有jar包,拷贝到eclipse中mapreduce3项目的hadoop2lib目录下。

选中hadoop2lib目录下所有jar包,单击右键,选择Build Path→Add to Build Path。

7.编写Java代码,并描述其设计思路

在MapReduce过程中默认就有对数据的排序。它是按照key值进行排序的,如果key为封装int的IntWritable类型,那么MapReduce会按照数字大小对key排序,如果Key为封装String的Text类型,那么MapReduce将按照数据字典顺序对字符排序。在本例中我们用到第一种,key设置为IntWritable类型,其中MapReduce程序主要分为Map部分和Reduce部分。

Map部分代码
view plain "copy

public static class Map extends Mapper<Object,Text,IntWritable,Text>{  private static Text goods=new Text();  private static IntWritable num=new IntWritable();  public void map(Object key,Text value,Context context) throws IOException, InterruptedException{  String line=value.toString();  String arr[]=line.split("\t");  num.set(Integer.parseInt(arr[1]));  goods.set(arr[0]);  context.write(num,goods);  }  }

在map端采用Hadoop默认的输入方式之后,将输入的value值用split()方法截取,把要排序的点击次数字段转化为IntWritable类型并设置为key,商品id字段设置为value,然后直接输出<key,value>。map输出的<key,value>先要经过shuffle过程把相同key值的所有value聚集起来形成<key,value-list>后交给reduce端。

Reduce部分代码
view plain "copy

public static class Reduce extends Reducer<IntWritable,Text,IntWritable,Text>{  private static IntWritable result= new IntWritable();  //声明对象result  public void reduce(IntWritable key,Iterable<Text> values,Context context) throws IOException, InterruptedException{  for(Text val:values){  context.write(key,val);  }  }  }

reduce端接收到<key,value-list>之后,将输入的key直接复制给输出的key,用for循环遍历value-list并将里面的元素设置为输出的value,然后将<key,value>逐一输出,根据value-list中元素的个数决定输出的次数。

完整代码
view plain "copy

package mapreduce;
import java.io.IOException;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class OneSort {  public static class Map extends Mapper<Object , Text , IntWritable,Text >{  private static Text goods=new Text();  private static IntWritable num=new IntWritable();  public void map(Object key,Text value,Context context) throws IOException, InterruptedException{  String line=value.toString();  String arr[]=line.split("\t");  num.set(Integer.parseInt(arr[1]));  goods.set(arr[0]);  context.write(num,goods);  }  }  public static class Reduce extends Reducer< IntWritable, Text, IntWritable, Text>{  private static IntWritable result= new IntWritable();  public void reduce(IntWritable key,Iterable<Text> values,Context context) throws IOException, InterruptedException{  for(Text val:values){  context.write(key,val);  }  }  }  public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException{  Configuration conf=new Configuration();  Job job =new Job(conf,"OneSort");  job.setJarByClass(OneSort.class);  job.setMapperClass(Map.class);  job.setReducerClass(Reduce.class);  job.setOutputKeyClass(IntWritable.class);  job.setOutputValueClass(Text.class);  job.setInputFormatClass(TextInputFormat.class);  job.setOutputFormatClass(TextOutputFormat.class);  Path in=new Path("hdfs://localhost:9000/mymapreduce3/in/goods_visit1");  Path out=new Path("hdfs://localhost:9000/mymapreduce3/out");  FileInputFormat.addInputPath(job,in);  FileOutputFormat.setOutputPath(job,out);  System.exit(job.waitForCompletion(true) ? 0 : 1);  }  }

8.在OneSort类文件中,右键并点击=>Run As=>Run on Hadoop选项,将MapReduce任务提交到Hadoop中。

9.待执行完毕后,进入命令模式下,在HDFS上/mymapreduce3/out中查看实验结果。
view plain "copy

hadoop fs -ls /mymapreduce3/out
hadoop fs -cat /mymapreduce3/out/part-r-00000

Mapreduce排序相关推荐

  1. mapreduce排序算法_MapReduce算法–二级排序

    mapreduce排序算法 我们将继续执行有关实现MapReduce算法的系列文章,该系列可在使用MapReduce进行数据密集型文本处理中找到. 本系列的其他文章: 使用MapReduce进行数据密 ...

  2. MapReduce排序-实现比较器和序列化代码

    要求: 第一列按照字典顺序进行排列 第一列相同的时候, 第二列按照升序进行排列 解决思路: 将 Map 端输出的 <key,value> 中的 key 和 value 组合成一个新的 ke ...

  3. MapReduce排序-概述

    MapReduce序列化和排序 序列化 (Serialization) 是指把结构化对象转化为字节流 反序列化 (Deserialization) 是序列化的逆过程. 把字节流转为结构化对象. 当要在 ...

  4. mapreduce排序比较器的选取

    排序比较器的选取顺序: Map: 取用户设置的排序比较器 取key自带的比较器 Reduce: 取用户设置的分组比较器 取用户设置的排序比较器 取key自带的比较器 转载于:https://www.c ...

  5. 大数据教程(9.1)流量汇总排序的mr实现

    2019独角兽企业重金招聘Python工程师标准>>> 上一章我们有讲到一个mapreduce案例--移动流量排序,如果我们要将最后的输出结果按总流量大小逆序输出,该怎么实现呢?本节 ...

  6. hadoop 实现数据排序

    前言 在很多业务场景下,需要对原始的数据读取分析后,将输出的结果按照指定的业务字段进行排序输出,方便上层应用对结果数据进行展示或使用,减少二次排序的成本 在hadoop的MapReduce中,提供了对 ...

  7. MapReduce剥洋葱

    一.大白话MapReduce 1.什么是Map/Reduce,看下面的各种解释: (1)MapReduce是hadoop的核心组件之一,hadoop要分布式包括两部分,一是分布式文件系统hdfs,一部 ...

  8. 什么是MapReduce?MapReduce整体架构搭建使用介绍

    文章目录 前言 MapReduce 入门 MapReduce的核心思想 MapReduce yarn Yarn伪分布式搭建 MapReduce编码 需求 MapReduce2.0工作机制 MapRed ...

  9. BigData CH4 MapReduce

    CH4 MapReduce 并行编程框架 并行计算 划分子任务.计算.合并结果 构抽象模型Map与Reduce Map 对一组数据元素进行某种重复式的处理 map: (k1; v1) → \right ...

最新文章

  1. Java并发编程实战笔记2:对象的组合
  2. 未来教育python视频百度云-青橙课程 | 人工智能走进课堂,为未来教育高质量发展赋能!...
  3. POJ 2336 Ferry Loading II 动态规划
  4. 数据中心人员短缺,行业仍然充满挑战
  5. 个人工作总结10(第二阶段)
  6. 百度地图标点点击变色_《和平精英》版本爆料第三弹:雪地洞穴开启!组队标点功能升级~...
  7. python+html语音人物交互_使用Python读取HTML表并与之交互
  8. require 动态加载_require,exports,module.exports和import,export,export default
  9. 雷林鹏分享:C# 多态性
  10. FontAwesome图标大全
  11. jQuery ajax get与post后台交互中的奥秘
  12. hdu 4300 Clairewd’s message kmp匹配! 多校联合赛第一题
  13. MagicHouse- 智能家居管理机器人 在线演示
  14. 【数据分享】历次人口普查数据(一普到七普)
  15. hdu 4609 3-idiots——FFT
  16. SRIO传输协议学习
  17. chm文件打不开的办法
  18. CSS的压缩 方法与解压
  19. 向上转型 (Upcasting)
  20. mid代表计算机,MID是什么

热门文章

  1. 大数据知识面试题-通用(2022版)
  2. Uva-755-487--3279
  3. 分布式机器学习(上)-并行计算与机器学习
  4. 面经手册 · 第3篇《HashMap核心知识,扰动函数、负载因子、扩容链表拆分,深度学习》
  5. 【蝴蝶算法】基于随机惯性权重策略+最优邻域扰动策略+动态转换概率策略的蝴蝶算法求解单目标优化问题附matlab代码IBOA...
  6. python3代码兼容python2
  7. QPushButton setIcon 图片不显示
  8. JavaScript错误调试与处理——错误调试_断点
  9. 永不气馁,用清晰目标换取成功结果 开利网络主题培训会议持续进行中
  10. linux mono jexus性能,docker运行jexus+mono爬坑记,