课程安排
Partitioner编程** 自定义排序编程** Combiner编程** 常见的MapReduce算法** ---------------------------加深拓展---------------------- Mapreduce原理及源码分析
Partitioner编程
Partitioner是partitioner的基类,如果需要定制partitioner也需要继承该类。 2. HashPartitioner是mapreduce的默认partitioner。计算方法是 which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks,得到当前的目的reducer。 3. (例子以jar形式运行)
排序和分组
在map和reduce阶段进行排序时,比较的是k2。v2是不参与排序比较的。如果要想让v2也进行排序,需要把k2和v2组装成新的类,作为k2,才能参与比较。
分组时也是按照k2进行比较的。
Combiners编程
每一个map可能会产生大量的输出,combiner的作用就是在map端对输出先做一次合并,以减少传输到reducer的数据量。
combiner最基本是实现本地key的归并,combiner具有类似本地的reduce功能。
如果不用combiner,那么,所有的结果都是reduce完成,效率会相对低下。使用combiner,先完成的map会在本地聚合,提升速度。
注意:Combiner的输出是Reducer的输入,如果Combiner是可插拔的,添加Combiner绝不能改变最终的计算结果。所以Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致,且不影响最终结果的场景。比如累加,最大值等。
Job执行流程图
&
&
1.每个map有一个环形内存缓冲区,用于存储任务的输出。默认大小100MB(io.sort.mb属性),一旦达到阀值0.8(io.sort.spill.percent),一个后台线程把内容写到(spill)磁盘的指定目录(mapred.local.dir)下的新建的一个溢出写文件。
2.写磁盘前,要partition,sort。如果有combiner,combine排序后数据。
3.等最后记录写完,合并全部溢出写文件为一个分区且排序的文件。
1.Reducer通过Http方式得到输出文件的分区。
2.TaskTracker为分区文件运行Reduce任务。复制阶段把Map输出复制到Reducer的内存或磁盘。一个Map任务完成,Reduce就开始复制输出。
3.排序阶段合并map输出。然后走Reduce阶段。
hadoop的压缩codec
Codec为压缩,解压缩的算法实现。
在Hadoop中,codec由CompressionCode的实现来表示。下面是一些实现:
MapReduce的输出进行压缩
输出的压缩属性:
MapReduce常见算法
l单词计数
l数据去重
l排序
lTop K
l选择
l投影
l分组
l多表连接
l单表关联
思考题
l如何使用计数器
lCombiner的作用是什么,应用场景是什么
lPartitioner的作用是什么,应用场景是什么
lShuffler的过程是什么
===================第四天笔记=============================
1.实现分区的步骤: 1.1先分析一下具体的业务逻辑,确定大概有多少个分区 1.2首先书写一个类,它要继承org.apache.hadoop.mapreduce.Partitioner这个类 1.3重写public int getPartition这个方法,根据具体逻辑,读数据库或者配置返回相同的数字 1.4在main方法中设置Partioner的类,job.setPartitionerClass(DataPartitioner.class); 1.5设置Reducer的数量,job.setNumReduceTasks(6); 2.排序MR默认是按key2进行排序的,如果想自定义排序规则,被排序的对象要实现WritableComparable接口,在compareTo方法中实现排序规则,然后将这个对象当做k2,即可完成排序 3.combiner的作用就是在map端对输出先做一次合并,以减少传输到reducer的数据量。 4.MR启动流程 start-mapred.sh --> hadoop-daemon.sh --> hadoop --> org.apache.hadoop.mapred.JobTracker Jobtracker调用顺序:main --> startTracker --> new JobTracker 在其构造方法中首先创建一个调度器,接着创建一个RPC的server(interTrackerServer)tasktracker会通过PRC机制与其通信 然后调用offerService方法对外提供服务,在offerService方法中启动RPC server,初始化jobtracker,调用taskScheduler的start方法 --> eagerTaskInitializationListener调用start方法, --> 调用jobInitManagerThread的start方法,因为其是一个线程,会调用JobInitManager的run方法 --> jobInitQueue任务队列去取第一个任务,然后把它丢入线程池中,然后调用-->InitJob的run方法 --> jobTracker的initJob方法 --> JobInProgress的initTasks --> maps = new TaskInProgress[numMapTasks]和reduces = new TaskInProgress[numReduceTasks]; TaskTracker调用顺序:main --> new TaskTracker在其构造方法中调用了initialize方法,在initialize方法中调用RPC.waitForProxy得到一个jobtracker的代理对象 接着TaskTracker调用了本身的run方法,--> offerService方法 --> transmitHeartBeat返回值是(HeartbeatResponse)是jobTracker的指令,在transmitHeartBeat方法中InterTrackerProtocol调用了heartbeat将tasktracker的状态通过RPC机制发送给jobTracker,返回值就是JobTracker的指令 heartbeatResponse.getActions()得到具体的指令,然后判断指令的具体类型,开始执行任务 addToTaskQueue启动类型的指令加入到队列当中,TaskLauncher又把任务加入到任务队列当中,--> TaskLauncher的run方法 --> startNewTask方法 --> localizeJob下载资源 --> launchTaskForJob开始加载任务 --> launchTask --> runner.start()启动线程; --> TaskRunner调用run方法 --> launchJvmAndWait启动java child进程