处理数据的输入路径—–计算任务切片(mt个数),根据任务切片的个数启动mapTask,MapTask, 处理属于自己的任务数据(start(偏移量),length) ,TextInputFormat就是数据切分,分成若干个split,以便确定MapTask个数,以及对应的split,且里面有一个createRecorder方法,返回一个LineRecordReader,一行一行的读(如果行数比较多,就会产生大量的kv),offset—>K linecontent —->v,map阶段读取的时候,虽然规定以128M结尾,但是不以标准的128M的字节就结束,假设是这样,万一刚好以在这个节点切开了摸一个单词,这样也会导致信息缺失,所以一般是在128M这个一行,以 \n 结束.
文件的个数至少是3个 ,假设有有300M ,128M Task0 128~256 Task1 剩下的44M>128
0.1 就划分到第三个Task里面去, 如果 剩下的内存小于128
0.1 就会把这部分的内容划分到 前面两个Task里面去’,如果mapTask是这样划分的,如果是一个普通文本的,那么是按照文本个数和文本大小进行划分,
String[] ws = v.toString().split(“\s+”)
(w,1)(w,1)(w,1)(w,1)(w,1)(w,1)(w,1)
也是把这些kv封装在了Text IntWritable 里面,a,1,a,1b,1,c,1,f,1
context.write(a,1)
context.write(a,1)
context.write(a,1)
HashPartion里面的getPartion(){
return(key.hashCode()&Integer.MAX_VALUE)%numReduceTasks} 注意并上一个Max_VALUE是为了取模的时候取到一个负数
产生大量的kv交给自定义的map类
map(k,v ,context) 一个kv就会执行一次map方法,map里面有一个SequenceInputFormat可以读取各种类型的文件,这也一样会生产kv(Sequence是首位相连的kv,kv…)
3.每对的kv—->map(key,value,context)方法,处理一行数据value,生产新的kv,
4.context.write(NK,NV)—–>MapOutPutBufffer (ctrl+ o进入里面的源码)
MapOutputBuffer是一个用来暂时存储map输出的缓冲区，它的缓冲区大小是有限的，当写入的数据超过缓冲区的设定的阀值时，需要将缓冲区的数据溢出写入到磁盘，这个过程称之为spill，spill的动作会通过Condition通知给SpillThread，由SpillThread完成具体的处理过程。如果缓冲区使用过的是简单的单向缓冲区，在一次写满后，flush到磁盘，那么在flush的过程中，将会严重影响到map向缓冲区写入的性能，因为在flush的时候，缓冲区是需要被锁定的。因此，MR采用了循环缓冲区，做到数据在spill的同时，仍然可以向剩余空间继续写入数据。
在MapOutputBuffer里面
collect(k key,v value,final int partion) 在经过map里面进行输出的时候(a,1-1) 或者(b,1-0) 0,1是取模得到的分区标识.便于分区,分区的个数也是根据reduce个数决定的.得到经过分区标识的kv进入到圆形缓冲区,kv,he 元数据kvmeta,分别以不同方向进入缓冲区,kv写到80%就停止往里面写内容,要留一点空间给元数据
在环形数组里对k进行排序,当到达80%的时候,就会出现数据的溢出spill(数据溢出组件)溢出得救写入达到磁盘
输出的时候a,1a,1b,1b,1,f,1f,1…这时候是区内排序,不是全局排序,也就是分区编号相同的才会进行一起排序因此得到的是
a,1,a1c,1,c1,e,1,e,1(假设按照2取模)
b,1,b,1,b1,d,1,d,1,
假设当前面80%的内容进行分区处理完后,还有剩下的,依然要根据之前的划分进行分区
这样就会差生多次溢出的区域内容,
对这些区再次进行合并,成一组,然后进行组内排序
a,1,a1c,1,c1,e,1,e,1|| b,1,b,1,b1,d,1,d,1,
然后输出 (TextOutFormat) 普通文本,因为开始读的是普通文本,
但是普通文本程序读起来很不方便,所以可以转化成首位相连的一窜kv输出流
称为SequenceFileOutputFormat
Remote Procedure Calls 远程过程调用 (RPC) ，程序可使用这种协议向网络中的另一台计算机上的程序请求服务。
reduce阶段从mapTask阶段处理好的数据下载也叫抓取(Facher),
因为有多个数据块,但是在不同的数据块根据一些规则进行分区的时候,会产相同的区号,这时候相同的区号就会通过shuffle 分配到同一reduce里面()重新再次合并(Merger),shuffle就像是一个重新洗牌分发牌的一个过程
当得到两个MapTask的两个同一个区号信息时,再次进行分组排序
里面会调用GroupingCompator 组件里面的方法compare(k1,k2)(判断两个相邻的key是否相同,如果相同存储在同一个等迭代器中交给同一个rudece方法处理)
a—–>Iterator<11111111>
f——>Iterator<11111111>显示的是每类字母的个数
实现聚合操作 输出a 8 f 7 …
在OuputFormat类,这个类里面有对应输出的路径,也是结果结果路径
这个结果路径可以说本地路径也可以是hdfs 路径
第一种比较笨的方法:从头开始读取数据,比如 1001,zss,28,M 28改成38,则可以根据底层追加的原
理,每当读到该数据,就在没填一个序列号,第一次读到添1,第二次读到添加2 读到一次就修改成我们想要的数据,并序列号i++ ,并覆盖到前面的数据
在最后追加到文件数据的末尾.
序列化 (Serialization)是将对象的状态信息转换为可以存储或传输的形式的过程。在序列化期间，对象将其当前状态写入到临时或持久性存储区。
namenode的作用
1.接收datanode的注册——>管理datanode的存储内容—–>分配datanode的存储任务—–>
2.接收客户端的上传下载的请求—–>返回相应的元数据信息—–>管理和维护数据存储的元数据信息——–>
3.管理datanode的心跳机制,—–>接收datanode的数据汇报
4.维护数据物理切块副本,给DN分配副本复制任务
datanode
真正存储数据的节点
处理客户端的读写任务
向NN注册, 汇报存储容量
接收NN的复制副本的任务,并维护集群中数据副本个数
namenode和datanode的通信机制
hdfs.site.xml文件里面
dfs-namenode-rpc-address=linux01://90000
checkpoint的存储机制
客户机在对namenode进行数据的上传或下载操作过程中,会产生大量的元数据信息,这些元数据信息是以树形的结果在内存中存储,树本身是一个类,为了保证万一机器宕机后避免这些元数据信息丢失,必须在磁盘里面还有备份一个相对应的文件该文件称为FsImage,如何传进去的呢,就是通过读写操作以二进制流的形式传输到FsImage中,要知道把内存里面的树形结构对象通过写的操作编程二进制流输出流,这个过程就是序列化,再把二进制流以读的形式输入到内存当中变成树形结构,这个过程就是反序列化.在这个要明白
如过内存里面没进行一次添加一次客户端的上传和下载的元数据信息,但是这些某段时间添加的元数据树不能及时传输到FsImage里面的,这样万一在这时候宕机了,镜像文件复原的元数据信息就会不全,这时候我们该怎么做呢
比如每一个小时把内存里面的信息加载到磁盘的镜像文件中,但这一个小时内,客户端对namenode传达的上传或下载的元数据信息,就只能在内存中,不能及时传输到镜像文件中,这样会出现问题,因此在这一个小时内就要添加日志来保存这一个小时内存都做了哪些操作 ,比如上传文件的大小,上传文件到那个机器,哪个具体位置.都用日志来保存.
这样万一机器一点到两点之间,本来两点的时候,内存就要进行一次更新传输到镜像文件新的内容的操作,而在一点五十就发生的宕机现象,这样镜像文件可以使磁盘回复一点以前的存储内容,一点到一点五十的元数据就要通过日志来恢复,这样在重启的时候,内存里面的内容依然能够恢复到宕机以前的状态,
如果这一个小时期间,机器做了很多的操作,日志记录的大量的元数据信息,如果宕机重启也会话费很长时间,这时候secondary 就其作用了,在机器初始化的时候,也就是,镜像文件为空的时候,把该文件复制到一份02号机器也就是secondary 里面,同时假设在一点进行重新内存和磁盘传输,同时那这些传输的信息,复制到Secondary一份包括后面的日志,也就是在一点五十宕机之前,secondary,和镜像文件里面的内容和01号的有模有样,这样在机器重启的时候就算发费很长时间,在这段时间里面又有新的内容加入到日志里面,这样我们直接把secondary的文件传输到一号机,并读到内存当中,而01号机只需要修改添加新添加的一些日志所显示的元数据信息,这样就会生不少时间.
MR处理的基本流程
