Hadoop生态圈基础（zookeeper,hdfs,yarn，MR,hbase,hive）_hadoop mr,hdfs,yarn hive

安装并配置zookeeper

tar -zxvf zookeeper-3.5.1-alpha.tar.gz

mv zookeeper-3.5.1-alpha zookeeper

cd zookeeper

mkdir data

mkdir log

cd conf

cp zoo_sample.cfg zoo.cfg

 

vi zoo.cfg

dataDir=/usr/local/zookeeper/data

dataLogDir=/usr/local/zookeeper/log

server.128=192.168.35.128:5881:6881

server.129=192.168.35.129:5881:6881

server.130=192.168.35.130:5881:6881

 

cd /usr/local/zookeeper/data

touch myid

vi myid

128/129/130(server后面的值，主要用来辨识机子)

 

cd /usr/local/zookeeper/bin

./zkServer.sh start                     执行zkServer二进制可执行文件启动zkServer服务

jps

./zkCli.sh -server192.168.35.128:2181   执行zkCli.sh二进制可执行文件打开zookeeper客户端

 

操作zookeeper

创建节点：create /test

获取内容：get /test

设置内容：set /test

删除：delete /test

查看状态：stat /test

Zookeeper工作原理（详细）

1. Zookeeper的角色

» 领导者（leader），负责进行投票的发起和决议，更新系统状态
» 学习者（learner），包括跟随者（follower）和观察者（observer），follower用于接受客户端请求并想客户端返回结果，在选主过程中参与投票
» Observer可以接受客户端连接，将写请求转发给leader，但observer不参加投票过程，只同步leader的状态，observer的目的是为了扩展系统，提高读取速度
» 客户端（client），请求发起方　　

• Zookeeper的核心是原子广播，这个机制保证了各个Server之间的同步。实现这个机制的协议叫做Zab协
  议。Zab协议有两种模式，它们分别是恢复模式（选主）和广播模式（同步）。当服务启动或者在领导者
　　　崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数Server完成了和leader的状态同步以后
 ，恢复模式就结束了。状态同步保证了leader和Server具有相同的系统状态。

• 为了保证事务的顺序一致性，zookeeper采用了递增的事务id号（zxid）来标识事务。所有的提议（
proposal）都在被提出的时候加上了zxid。实现中zxid是一个64位的数字，它高32位是epoch用来标识
  leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch，标识当前属于那个leader的
　　　统治时期。低32位用于递增计数。
• 每个Server在工作过程中有三种状态：
LOOKING：当前Server不知道leader是谁，正在搜寻
LEADING：当前Server即为选举出来的leader
FOLLOWING：leader已经选举出来，当前Server与之同步

2、Zookeeper 的读写机制

» Zookeeper是一个由多个server组成的集群
» 一个leader，多个follower
» 每个server保存一份数据副本
» 全局数据一致
» 分布式读写
» 更新请求转发，由leader实施

 

3、Zookeeper 的保证　

» 更新请求顺序进行，来自同一个client的更新请求按其发送顺序依次执行
» 数据更新原子性，一次数据更新要么成功，要么失败
» 全局唯一数据视图，client无论连接到哪个server，数据视图都是一致的
» 实时性，在一定事件范围内，client能读到最新数据

4、Zookeeper节点数据操作流程

　　　　注：1.在Client向Follwer发出一个写的请求

2.Follwer把请求发送给Leader

3.Leader接收到以后开始发起投票并通知Follwer进行投票

4.Follwer把投票结果发送给Leader

5.Leader将结果汇总后如果需要写入，则开始写入同时把写入操作通知给Leader，然后commit;

6.Follwer把请求结果返回给Client

• Follower主要有四个功能：
• 1. 向Leader发送请求（PING消息、REQUEST消息、ACK消息、REVALIDATE消息）；
• 2 .接收Leader消息并进行处理；
• 3 .接收Client的请求，如果为写请求，发送给Leader进行投票；
• 4 .返回Client结果。
• Follower的消息循环处理如下几种来自Leader的消息：
• 1 .PING消息： 心跳消息；
• 2 .PROPOSAL消息：Leader发起的提案，要求Follower投票；
• 3 .COMMIT消息：服务器端最新一次提案的信息；
• 4 .UPTODATE消息：表明同步完成；
• 5 .REVALIDATE消息：根据Leader的REVALIDATE结果，关闭待revalidate的session还是允许其接受消息；
• 6 .SYNC消息：返回SYNC结果到客户端，这个消息最初由客户端发起，用来强制得到最新的更新。

5、Zookeeper leader 选举　

• 半数通过
– 3台机器 挂一台 2>3/2
– 4台机器 挂2台 2！>4/2

• A提案说，我要选自己，B你同意吗？C你同意吗？B说，我同意选A；C说，我同意选A。(注意，这里超过半数了，其实在现实世界选举已经成功了。

　　　但是计算机世界是很严格，另外要理解算法，要继续模拟下去。)
• 接着B提案说，我要选自己，A你同意吗；A说，我已经超半数同意当选，你的提案无效；C说，A已经超半数同意当选，B提案无效。
• 接着C提案说，我要选自己，A你同意吗；A说，我已经超半数同意当选，你的提案无效；B说，A已经超半数同意当选，C的提案无效。
• 选举已经产生了Leader，后面的都是follower，只能服从Leader的命令。而且这里还有个小细节，就是其实谁先启动谁当头。

6、zxid

• znode节点的状态信息中包含czxid, 那么什么是zxid呢?
• ZooKeeper状态的每一次改变, 都对应着一个递增的Transaction id, 该id称为zxid. 由于zxid的递增性质, 如果zxid1小于zxid2, 那么zxid1肯定先于zxid2发生.

　　　创建任意节点, 或者更新任意节点的数据, 或者删除任意节点, 都会导致Zookeeper状态发生改变, 从而导致zxid的值增加.

7、Zookeeper工作原理

» Zookeeper的核心是原子广播，这个机制保证了各个server之间的同步。实现这个机制的协议叫做Zab协议。Zab协议有两种模式，它们分别是恢复模式和广播模式。

　　　当服务启动或者在领导者崩溃后，Zab就进入了恢复模式，当领导者被选举出来，且大多数server的完成了和leader的状态同步以后，恢复模式就结束了。

　　　状态同步保证了leader和server具有相同的系统状态

» 一旦leader已经和多数的follower进行了状态同步后，他就可以开始广播消息了，即进入广播状态。这时候当一个server加入zookeeper服务中，它会在恢复模式下启动，

　　　发现leader，并和leader进行状态同步。待到同步结束，它也参与消息广播。Zookeeper服务一直维持在Broadcast状态，直到leader崩溃了或者leader失去了大部分

　　　的followers支持。

» 广播模式需要保证proposal被按顺序处理，因此zk采用了递增的事务id号(zxid)来保证。所有的提议(proposal)都在被提出的时候加上了zxid。

　　　实现中zxid是一个64为的数字，它高32位是epoch用来标识leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch。低32位是个递增计数。

» 当leader崩溃或者leader失去大多数的follower，这时候zk进入恢复模式，恢复模式需要重新选举出一个新的leader，让所有的server都恢复到一个正确的状态。　

» 每个Server启动以后都询问其它的Server它要投票给谁。
» 对于其他server的询问，server每次根据自己的状态都回复自己推荐的leader的id和上一次处理事务的zxid（系统启动时每个server都会推荐自己）
» 收到所有Server回复以后，就计算出zxid最大的哪个Server，并将这个Server相关信息设置成下一次要投票的Server。
» 计算这过程中获得票数最多的的sever为获胜者，如果获胜者的票数超过半数，则改server被选为leader。否则，继续这个过程，直到leader被选举出来　　

» leader就会开始等待server连接
» Follower连接leader，将最大的zxid发送给leader
» Leader根据follower的zxid确定同步点
» 完成同步后通知follower 已经成为uptodate状态
» Follower收到uptodate消息后，又可以重新接受client的请求进行服务了

8、数据一致性与paxos 算法

• 据说Paxos算法的难理解与算法的知名度一样令人敬仰，所以我们先看如何保持数据的一致性，这里有个原则就是：
• 在一个分布式数据库系统中，如果各节点的初始状态一致，每个节点都执行相同的操作序列，那么他们最后能得到一个一致的状态。
• Paxos算法解决的什么问题呢，解决的就是保证每个节点执行相同的操作序列。好吧，这还不简单，master维护一个
  全局写队列，所有写操作都必须 放入这个队列编号，那么无论我们写多少个节点，只要写操作是按编号来的，就能保证一
　　　致性。没错，就是这样，可是如果master挂了呢。
• Paxos算法通过投票来对写操作进行全局编号，同一时刻，只有一个写操作被批准，同时并发的写操作要去争取选票，
　　　只有获得过半数选票的写操作才会被 批准（所以永远只会有一个写操作得到批准），其他的写操作竞争失败只好再发起一
　　　轮投票，就这样，在日复一日年复一年的投票中，所有写操作都被严格编号排 序。编号严格递增，当一个节点接受了一个
　　　编号为100的写操作，之后又接受到编号为99的写操作（因为网络延迟等很多不可预见原因），它马上能意识到自己 数据
　　　不一致了，自动停止对外服务并重启同步过程。任何一个节点挂掉都不会影响整个集群的数据一致性（总2n+1台，除非挂掉大于n台）。
 总结
• Zookeeper 作为 Hadoop 项目中的一个子项目，是 Hadoop 集群管理的一个必不可少的模块，它主要用来控制集群中的数据，

　　　如它管理 Hadoop 集群中的 NameNode，还有 Hbase 中 Master Election、Server 之间状态同步等。\
9、Observer

• Zookeeper需保证高可用和强一致性；
• 为了支持更多的客户端，需要增加更多Server；
• Server增多，投票阶段延迟增大，影响性能；
• 权衡伸缩性和高吞吐率，引入Observer
• Observer不参与投票；
• Observers接受客户端的连接，并将写请求转发给leader节点；
• 加入更多Observer节点，提高伸缩性，同时不影响吞吐率

10、 为什么zookeeper集群的数目，一般为奇数个？

•Leader选举算法采用了Paxos协议；
•Paxos核心思想：当多数Server写成功，则任务数据写成功如果有3个Server，则两个写成功即可；如果有4或5个Server，则三个写成功即可。
•Server数目一般为奇数（3、5、7）如果有3个Server，则最多允许1个Server挂掉；如果有4个Server，则同样最多允许1个Server挂掉由此，

 我们看出3台服务器和4台服务器的的容灾能力是一样的，所以为了节省服务器资源，一般我们采用奇数个数，作为服务器部署个数。

11、Zookeeper 的数据模型　

» 层次化的目录结构，命名符合常规文件系统规范
» 每个节点在zookeeper中叫做znode,并且其有一个唯一的路径标识
» 节点Znode可以包含数据和子节点，但是EPHEMERAL类型的节点不能有子节点
» Znode中的数据可以有多个版本，比如某一个路径下存有多个数据版本，那么查询这个路径下的数据就需要带上版本
» 客户端应用可以在节点上设置监视器
» 节点不支持部分读写，而是一次性完整读写

12、Zookeeper 的节点

» Znode有两种类型，短暂的（ephemeral）和持久的（persistent）
» Znode的类型在创建时确定并且之后不能再修改
» 短暂znode的客户端会话结束时，zookeeper会将该短暂znode删除，短暂znode不可以有子节点
» 持久znode不依赖于客户端会话，只有当客户端明确要删除该持久znode时才会被删除
» Znode有四种形式的目录节点
» PERSISTENT（持久的）
» EPHEMERAL(暂时的)
» PERSISTENT_SEQUENTIAL（持久化顺序编号目录节点）
» EPHEMERAL_SEQUENTIAL（暂时化顺序编号目录节点）

 

ZooKeeper 典型的应用场景

Zookeeper 从设计模式角度来看，是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper 就将负责通知已经在 Zookeeper 上注册的那些观察者做出相应的反应，从而实现集群中类似 Master/Slave 管理模式

 

Flume

cd /usr/local/flume/conf

vi flume2kafka.conf

 

a1.sources = r1

a1.sinks = k1

a1.channels = c1

#监测池

a1.sources.r1.type = spooldir

a1.sources.r1.spoolDir = /home/majiayu/tmp/flumetokafka/mjy    

 

a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink

a1.sinks.k1.topic = bigdata_3

a1.sinks.k1.brokerList = 192.168.17.128:9092,192.168.17.129:9092,192.168.17.130:9092

a1.sinks.k1.requiredAcks = 1

a1.sinks.k1.batchSize = 100

 

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

 

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

 

执行监测

bin/flume-ng agent -n a1 -c conf -f conf/flume2kafka.conf

 

Flume应用场景：将日志数据信息收集并移交给Hadoop平台上去分析.

Flume的结构

1. flume的外部结构：

   

数据发生器（如：facebook,twitter）产生的数据被单个的运行在数据发生器所在服务器上的agent所收集，之后数据收容器从各个agent上汇集数据并将采集到的数据存入到HDFS或者HBase中

 

2. Flume 事件

  事件作为Flume内部数据传输的最基本单元.

 

3.Flume Agent

  我们在了解了Flume的外部结构之后,知道了Flume内部有一个或者多个Agent,然而对于每一个Agent来说,它就是一共独立的守护进程(JVM),它从客户端哪儿接收收集,或者从其他的 Agent哪儿接收,然后迅速的将获取的数据传给下一个目的节点sink,或者agent. 如下图所示flume的基本模型

 

Agent主要由:source,channel,sink三个组件组成.

Source:               接受数据，类型有多种。

   从数据发生器接收数据,并将接收的数据以Flume的event格式传递给一个或者多个通道channal,Flume提供多种数据接收的方式,比如Avro,Thrift,twitter1%等

Channel:         临时存放地，对Source中来的数据进行缓冲，直到sink消费掉。

 channal是一种短暂的存储容器,它将从source处接收到的event格式的数据缓存起来,直到它们被sinks消费掉,它在source和sink间起着一共桥梁的作用,channal是一个完整的事务,这一点保证了数据在收发的时候的一致性. 并且它可以和任意数量的source和sink链接. 支持的类型有: JDBC channel , File System channel , Memort channel等.

sink:         从channel提取数据存放到中央化存储(hadoop / hbase)。

  sink将数据存储到集中存储器比如Hbase和HDFS,它从channals消费数据(events)并将其传递给目标地. 目标地可能是另一个sink,也可能HDFS,HBase.

 

它的组合形式举例:

 

以上介绍的flume的主要组件,下面介绍一下Flume插件:

1. Interceptors拦截器

   用于source和channel之间,用来更改或者检查Flume的events数据

2. 管道选择器 channels Selectors

   在多管道是被用来选择使用那一条管道来传递数据(events). 管道选择器又分为如下两种:

   默认管道选择器:  每一个管道传递的都是相同的events

  多路复用通道选择器:  依据每一个event的头部header的地址选择管道.

3.sink线程

   用于激活被选择的sinks群中特定的sink,用于负载均衡.

 

 

 

 

Hadoop HDFS安装配置及基本操作

1, 解压tar -zxvf haoop.tar.gz

2, service iptables stop

3, 重命名mv hadoop-2.6.1 hadoop_2_6

4,配置环境变量 /etc/profile

export HADOOP_HOME=/usr/local/hadoop_2_6

export PATH=$PATH:$HADOOP_HOME/bin

5, source /etc/profile

6,配置hadoop-env.sh   在/usr/local/hadoop_2_6/etc/hadoop  

export JAVA_HOME=/usr/local/jdk1_8

7, core-site.xml  在/usr/local/hadoop_2_6/etc/hadoop  

<configuration>

    <property>

        <name>hadoop.tmp.dir</name>

        <value>file:/home/wangaibings/hadoop/data</value>

    </property>

    <property>

        <name>fs.defaultFS</name>

        <value>hdfs://192.168.119.131:9000</value>

    </property>

   <!-- <property>

        <name>io.file.buffer.size</name>

        <value>131702</value>

    </property>-->

</configuration>

8，hdfs-site.xml   在/usr/local/hadoop_2_6/etc/hadoop

<configuration>

    <property>

             <name>dfs.replication</name>

             <value>1</value>

     </property>

     <property>

             <name>dfs.namenode.name.dir</name>

             <value>file:/home/wangaibing/hadoop/hdfs/name</value>

      </property>

     <property>

             <name>dfs.datanode.data.dir</name>

             <value>file:/home/wangaibing/hadoop/hdfs/data</value>

      </property>

</configuration>

9, 配置masters, slaves在/usr/local/hadoop_2_6/etc/hadoop下(没有就新建)

[root@data1 hadoop]# vi slaves

[root@data1 hadoop]# cat slaves

192.168.17.128

[root@data1 hadoop]# vi masters

[root@data1 hadoop]# cat masters       192.168.17.128

10, Hadoop启动 (在这边只使用hdfs,不用yarn或是map)

    1）格式化namenode

     #bin/hdfs namenode -format  (not hadoop namenode -format)

    2）启动NameNode 和 DataNode 守护进程

   接下来就可以启动进程了(通过jps查看是否启动成功)，

分别要启动namenode、datanode和secondarynamenode，分别执行一下命令：

./sbin/hadoop-daemon.sh start namenode

./sbin/hadoop-daemon.sh start datanode

./sbin/hadoop-daemon.sh start secondarynamenode

 

cd /usr/local/sbin

./stop-dfs.sh  一起停

./start-dfs.sh 一起开

 

[root@data1 bin]# pwd

/usr/local/hadoop_2_6/bin

[root@data1 bin]# ls

container-executor  hdfs      mapred.cmd               yarn

hadoop              hdfs.cmd  rcc                      yarn.cmd

hadoop.cmd          mapred    test-container-executor

[root@data1 bin]#

 

hadoop一些命令（在/usr/local/hadoop_2_6/bin目录下面执行)

HADOOP1.X中HDFS工作原理

 

HDFS（Hadoop Distributed File System ）Hadoop分布式文件系统。是根据google发表的论文翻版的。论文为GFS（Google File System）Google 文件系统（中文，英文）。

HDFS有很多特点：

    ① 保存多个副本，且提供容错机制，副本丢失或宕机自动恢复。默认存3份。

    ② 运行在廉价的机器上。（商用机）

    ③ 适合大数据的处理。多大？多小？HDFS默认会将文件分割成block，64M为1个block。然后将block按键值对存储在HDFS上，并将键值对的映射存到内存中。如果小文件太多，那内存的负担会很重。

如图所示，HDFS也是按照Master和Slave的结构。分NameNode、SecondaryNameNode、DataNode这几个角色。

NameNode：是Master节点，是大领导。管理数据块映射；处理客户端的读写请求；配置副本策略；管理HDFS的名称空间；

SecondaryNameNode：是一个小弟，分担大哥namenode的工作量；是NameNode的冷备份；合并fsimage和fsedits然后再发给namenode。

DataNode：Slave节点，奴隶，干活的。负责存储client发来的数据块block；执行数据块的读写操作。

热备份：b是a的热备份，如果a坏掉。那么b马上运行代替a的工作。冷备份：b是a的冷备份，如果a坏掉。那么b不能马上代替a工作。但是b上存储a的一些信息，减少a坏掉之后的损失。

fsimage:元数据镜像文件（文件系统的目录树。）

edits：元数据的操作日志（针对文件系统做的修改操作记录）

namenode内存中存储的是=fsimage+edits。

SecondaryNameNode负责定时默认1小时，从namenode上，获取fsimage和edits来进行合并，然后再发送给namenode。减少namenode的工作量。所以讲secondarynamenode，单独放置到一台机器上，可以增大冗余，但是有可能会丢失一小时内处理的数据。

工作原理

写操作：(有图)

有一个文件FileA，100M大小。Client将FileA写入到HDFS上。

HDFS按默认配置。

HDFS分布在三个机架上Rack1，Rack2，Rack3。

 

a. Client将FileA按64M分块。分成两块，block1和Block2;

b. Client向nameNode发送写数据请求，如图蓝色虚线①——>。

c. NameNode节点，记录block信息。并返回可用的DataNode，如粉色虚线②———>。

     Block1: host2,host1,host3

     Block2: host7,host8,host4

    原理：

        NameNode具有RackAware机架感知功能，这个可以配置。

         若client为DataNode节点，那存储block时，规则为：副本1，同client的节点上；副本2，不同机架节点上；副本3，同第二个副本机架的另一个节点上；其他副本随机挑选。

         若client不为DataNode节点，那存储block时，规则为：副本1，随机选择一个节点上；副本2，不同副本1，机架上；副本3，同副本2相同的另一个节点上；其他副本随机挑选。

d. client向DataNode发送block1；发送过程是以流式写入。

   流式写入过程，

     1>将64M的block1按64k的package划分;

     2>然后将第一个package发送给host2;

     3>host2接收完后，将第一个package发送给host1，同时client想host2发送第二个package；

     4>host1接收完第一个package后，发送给host3，同时接收host2发来的第二个package。

     5>以此类推，如图红线实线所示，直到将block1发送完毕。

     6>host2,host1,host3向NameNode，host2向Client发送通知，说“消息发送完了”。如图粉红颜色实线所示。

     7>client收到host2发来的消息后，向namenode发送消息，说我写完了。这样就真完成了。如图黄色粗实线

     8>发送完block1后，再向host7，host8，host4发送block2，如图蓝色实线所示。

     9>发送完block2后，host7,host8,host4向NameNode，host7向Client发送通知，如图浅绿色实线所示。

     10>client向NameNode发送消息，说我写完了，如图黄色粗实线。。。这样就完毕了。

分析，通过写过程，我们可以了解到：

        ①写1T文件，我们需要3T的存储，3T的网络流量贷款。

        ②在执行读或写的过程中，NameNode和DataNode通过HeartBeat进行保存通信，确定DataNode活着。如果发现DataNode死掉了，就将死掉的DataNode上的数据，放到其他节点去。读取时，要读其他节点去。

        ③挂掉一个节点，没关系，还有其他节点可以备份；甚至，挂掉某一个机架，也没关系；其他机架上，也有备份。

 

读操作：

读操作就简单一些了，如图所示，client要从datanode上，读取FileA。而FileA由block1和block2组成。 

那么，读操作流程为：

a. client向namenode发送读请求。

b. namenode查看Metadata信息，返回fileA的block的位置。

     block1:host2,host1,host3

     block2:host7,host8,host4

c. block的位置是有先后顺序的，先读block1，再读block2。而且block1去host2上读取；然后block2，去host7上读取；

上面例子中，client位于机架外，那么如果client位于机架内某个DataNode上，例如,client是host6。那么读取的时候，遵循的规律是：

优选读取本机架上的数据。

注意，看了hdfs的布局，以及作用，这里需要考虑几个问题：

1、既然NameNode，存储小文件不太合适，那小文件如何处理？

2、NameNode在内存中存储了meta等信息，那么内存的瓶颈如何解决？

3、Secondary是NameNode的冷备份，那么SecondaryNamenode和Namenode不应该放到一台设备上，因为Namenode宕掉之后，SecondaryNamenode一般也就死了，那讲SecondaryNameNode放到其他机器上，如何配置？

4、NameNode宕机后，如何利用secondaryNameNode上面的备份的数据，恢复Namenode？

5、设备宕机，那么，文件的replication备份数目，就会小于配置值，那么该怎么办？

 

Hadoop集群搭建

192.168.17.128:2181  zookeeper

192.168.17.128:50070  hadoop网站

 

 

 

YARN基本框架介绍

1.YARN模块介绍

概述

YARN是一个资源管理、任务调度的框架，主要包含三大模块：ResourceManager（RM）、NodeManager（NM）、ApplicationMaster（AM）。其中，ResourceManager负责所有资源的监控、分配和管理；ApplicationMaster负责每一个具体应用程序的调度和协调；NodeManager负责每一个节点的维护。对于所有的applications，RM拥有绝对的控制权和对资源的分配权。而每个AM则会和RM协商资源，同时和NodeManager通信来执行和监控task。几个模块之间的关系如图所示。

ResourceManager

• ResourceManager负责整个集群的资源管理和分配，是一个全局的资源管理系统。
• NodeManager以心跳的方式向ResourceManager汇报资源使用情况（目前主要是CPU和内存的使用情况）。RM只接受NM的资源回报信息，对于具体的资源处理则交给NM自己处理。
• YARN Scheduler根据application的请求为其分配资源，不负责application job的监控、追踪、运行状态反馈、启动等工作。

NodeManager

• NodeManager是每个节点上的资源和任务管理器，它是管理这台机器的代理，负责该节点程序的运行，以及该节点资源的管理和监控。YARN集群每个节点都运行一个NodeManager。
• NodeManager定时向ResourceManager汇报本节点资源（CPU、内存）的使用情况和Container的运行状态。当ResourceManager宕机时NodeManager自动连接RM备用节点。
• NodeManager接收并处理来自ApplicationMaster的Container启动、停止等各种请求。

ApplicationMaster

• 用户提交的每个应用程序均包含一个ApplicationMaster，它可以运行在ResourceManager以外的机器上。
• 负责与RM调度器协商以获取资源（用Container表示）。
• 将得到的任务进一步分配给内部的任务(资源的二次分配)。
• 与NM通信以启动/停止任务。
• 监控所有任务运行状态，并在任务运行失败时重新为任务申请资源以重启任务。
• 当前YARN自带了两个ApplicationMaster实现，一个是用于演示AM编写方法的实例程序DistributedShell，它可以申请一定数目的Container以并行运行一个Shell命令或者Shell脚本；另一个是运行MapReduce应用程序的AM—MRAppMaster。
• 注：RM只负责监控AM，并在AM运行失败时候启动它。RM不负责AM内部任务的容错，任务的容错由AM完成。

2.YARN运行流程

 

1. client向RM提交应用程序，其中包括启动该应用的ApplicationMaster的必须信息，例如ApplicationMaster程序、启动ApplicationMaster的命令、用户程序等。
2. ResourceManager启动一个container用于运行ApplicationMaster。
3. 启动中的ApplicationMaster向ResourceManager注册自己，启动成功后与RM保持心跳。
4. ApplicationMaster向ResourceManager发送请求，申请相应数目的container。
5. ResourceManager返回ApplicationMaster的申请的containers信息。申请成功的container，由ApplicationMaster进行初始化。container的启动信息初始化后，AM与对应的NodeManager通信，要求NM启动container。AM与NM保持心跳，从而对NM上运行的任务进行监控和管理。
6. container运行期间，ApplicationMaster对container进行监控。container通过RPC协议向对应的AM汇报自己的进度和状态等信息。
7. 应用运行期间，client直接与AM通信获取应用的状态、进度更新等信息。
8. 应用运行结束后，ApplicationMaster向ResourceManager注销自己，并允许属于它的container被收回。

 基本术语

正式介绍container启动流程之前，先介绍几个术语：

（1）Application 用户提交的任何一个应用程序，在YARN中被称为Application。

（2）Container 一个Application通常会被分解成多个任务并行执行，其中，每个任务要使用一定量的资源，这些资源被封装成container。详细说来，container不仅包含一个任务的资源说明，还包含很多其他信息，比如Container对应的节点、启动container所需的文件资源、环境变量和命令等信息。

（3）资源本地化 在container中启动任务之前，先要为任务准备好各种文件资源，这些文件资源通常在用户提交应用程序时已上传到HDFS上，而container启动之前，需要下载到本地工作录下，该过程称为资源本地化。

MapReduce 工作原理

一、MapReduce作业运行流程
1.在客户端启动一个作业。

2.向JobTracker请求一个Job ID。

3.将运行作业所需要的资源文件复制到HDFS上，包括MapReduce程序打包的JAR文件、配置文件和客户端计算所得的输入划分信息。这些文件都存放在JobTracker专门为该作业创建的文件夹中。文件夹名为该作业的Job ID。JAR文件默认会有10个副本（mapred.submit.replication属性控制）；输入划分信息告诉了JobTracker应该为这个作业启动多少个map任务等信息。

4.JobTracker接收到作业后，将其放在一个作业队列里，等待作业调度器对其进行调度（这里是不是很像微机中的进程调度呢，呵呵），当作业调度器根据自己的调度算法调度到该作业时，会根据输入划分信息为每个划分创建一个map任务，并将map任务分配给TaskTracker执行。对于map和reduce任务，TaskTracker根据主机核的数量和内存的大小有固定数量的map槽和reduce槽。这里需要强调的是：map任务不是随随便便地分配给某个TaskTracker的，这里有个概念叫：数据本地化（Data-Local）。意思是：将map任务分配给含有该map处理的数据块的TaskTracker上，同时将程序JAR包复制到该TaskTracker上来运行，这叫“运算移动，数据不移动”。而分配reduce任务时并不考虑数据本地化。

5.TaskTracker每隔一段时间会给JobTracker发送一个心跳，告诉JobTracker它依然在运行，同时心跳中还携带着很多的信息，比如当前map任务完成的进度等信息。当JobTracker收到作业的最后一个任务完成信息时，便把该作业设置成“成功”。当JobClient查询状态时，它将得知任务已完成，便显示一条消息给用户。

以上是在客户端、JobTracker、TaskTracker的层次来分析MapReduce的工作原理的，下面我们再细致一点，从map任务和reduce任务的层次来分析分析吧。
二、Map、Reduce任务中Shuffle和排序的过程 

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操作，目的有两个：1.尽量减少每次写入磁盘的数据量；2.尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。为了减少网络传输的数据量，这里可以将数据压缩，只要将mapred.compress.map.out设置为true就可以了。

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

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

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执行过程+统计单词数例子

 

 

 

 

package mapreduce2;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

 

public class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

//key2 表示该行中的单词

final Text key2 = new Text();

//value2 表示单词在该行中的出现次数

final IntWritable value2 = new IntWritable(1);

 

//key 表示文本行的起始位置

//value 表示文本行

protected void map(LongWritable key, Text value, Context context) throws java.io.IOException, InterruptedException {

final String[] splited = value.toString().split(" ");

for (String word : splited) {

key2.set(word);

//把key2、value2写入到context中

context.write(key2, value2);

}

};

}

 

 

 

 

 

 

package mapreduce2;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

//value3表示单词出现的总次数

final IntWritable value3 = new IntWritable(0);

 

/**

* key 表示单词

* values 表示map方法输出的1的集合

* context 上下文对象

*/

protected void reduce(Text key, java.lang.Iterable<IntWritable> values, Context context) throws java.io.IOException, InterruptedException {

int sum = 0;

for (IntWritable count : values) {

sum += count.get();

}

//执行到这里，sum表示该单词出现的总次数

//key3表示单词，是最后输出的key

final Text key3 = key;

//value3表示单词出现的总次数，是最后输出的value

value3.set(sum);

context.write(key3, value3);

};

}

 

Reducer 类的四个泛型依次是<k2,v2,k3,v3>，要注意 reduce 方法的第二个参数是 java.lang.Iterable 类型，迭代的是 v2。也就是 k2 相同的 v2 都可以迭代出来。

驱动代码，如下

public class WordCountApp {

public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

//输入路径

final String INPUT_PATH = "hdfs://hadoop:9000/word.txt";

//输出路径，必须是不存在的

final String OUTPUT_PATH = "hdfs://hadoop:9000/output4";

//创建一个job对象，封装运行时需要的所有信息

final Job job = new Job(new Configuration(), "WordCountApp");

//如果需要打成jar运行，需要下面这句

//job.setJarByClass(WordCountApp.class);

//告诉job执行作业时输入文件的路径

FileInputFormat.setInputPaths(job, INPUT_PATH);

//设置把输入文件处理成键值对的类

job.setInputFormatClass(TextInputFormat.class);

//设置自定义的Mapper类

job.setMapperClass(MyMapper.class);

//设置map方法输出的k2、v2的类型

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(IntWritable.class);

//设置对k2分区的类

job.setPartitionerClass(HashPartitioner.class);

//设置运行的Reducer任务的数量

job.setNumReduceTasks(1);

//设置自定义的Reducer类

job.setReducerClass(MyReducer.class);

//设置reduce方法输出的k3、v3的类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

//告诉job执行作业时的输出路径

FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));

//指明输出的k3类型

job.setOutputKeyClass(Text.class);

//指明输出的v3类型

job.setOutputValueClass(IntWritable.class);

//让作业运行，直到运行结束，程序退出

job.waitForCompletion(true);

}在以上代码中，我们创建了一个 job 对象，这个对象封装了我们的任务，可以提交到Hadoop 独立运行

}最后一句 job.waitForCompletion(true)，表示把 job 对象提交给 Hadoop 运行，直到作业运行结束后才可以。

(MR)---使用groupingcomparator求同一订单中最大金额的订单

 

 

现在需要求出每一个订单中成交金额最大的一笔交易

分析：

1、利用“订单id和成交金额”作为key，可以将map阶段读取到的所有订单数据按照id分区，按照金额排序，发送到reduce

2、在reduce端利用groupingcomparator将订单id相同的kv聚合成组，然后取第一个即是最大值 

 

调用reduce用来对数据进行分组

//利用reduce端的GroupingComparator来实现将一组bean看成相同的key

public class ItemidGroupingComparator extends WritableComparator {

//传入作为key的bean的class类型，以及制定需要让框架做反射获取实例对象

protected ItemidGroupingComparator() {

super(OrderBean.class, true);

}

@Override

public int compare(WritableComparable a, WritableComparable b) {

OrderBean abean = (OrderBean) a;

OrderBean bbean = (OrderBean) b;

//比较两个bean时，指定只比较bean中的orderid,orderid相同的就分到同一组

return abean.getItemid().compareTo(bbean.getItemid());

}

}

 

 

Partitioner用来分区

public class ItemIdPartitioner extends Partitioner<OrderBean, NullWritable>{

@Override

public int getPartition(OrderBean bean, NullWritable value, int numReduceTasks) {

//相同id的订单bean，会发往相同的partition

//而且，产生的分区数，是会跟用户设置的reduce task数保持一致

return (bean.getItemid().hashCode() & Integer.MAX_VALUE) % numReduceTasks;

}

}

 

 

 

 

WritableComparable compareTo用来排序

public class OrderBean implements WritableComparable<OrderBean>{

private Text itemid;

private DoubleWritable amount;

public OrderBean() {}

public OrderBean(Text itemid, DoubleWritable amount) {

set(itemid, amount); }

public void set(Text itemid, DoubleWritable amount) {

   this.itemid = itemid;

this.amount = amount;}

public Text getItemid() {

return itemid;   }

public DoubleWritable getAmount() {

return amount;  }

@Override

public int compareTo(OrderBean o) {

int cmp = this.itemid.compareTo(o.getItemid());

if (cmp == 0) {

cmp = -this.amount.compareTo(o.getAmount());}

return cmp ;   }

@Override

public void write(DataOutput out) throws IOException {

out.writeUTF(itemid.toString());

out.writeDouble(amount.get()); }

@Override

public void readFields(DataInput in) throws IOException {

String readUTF = in.readUTF();

double readDouble = in.readDouble();

this.itemid = new Text(readUTF);

this.amount= new DoubleWritable(readDouble);   }

@Override

public String toString() {

return itemid.toString() + "\t" + amount.get();

}

}

主程序

  job.setNumReduceTasks(2);

  job.waitForCompletion(true);

}

}

 

 

 

 

 

 

 

 

 

 

MapReduce实现寻找共同好友

1.需求

给出A-O个人中每个人的好友列表，求出哪些人两两之间有共同好友，以及他们的共同好友都有谁.

原始数据：

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

输出格式     A-B: C,E

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

4.输出

1）第一阶段输出文件的内容

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

1. 第二阶段输出文件的内容

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

解析mr的shuffle

图一

清楚Shuffle的大致范围就成——怎样把map task的输出结果有效地传送到reduce端。也可以这样理解， Shuffle描述着数据从map task输出到reduce task输入的这段过程。 

        在Hadoop这样的集群环境中，大部分map task与reduce task的执行是在不同的节点上。当然很多情况下Reduce执行时需要跨节点去拉取其它节点上的map task结果。如果集群正在运行的job有很多，那么task的正常执行对集群内部的网络资源消耗会很严重。这种网络消耗是正常的，我们不能限制，能做的就是最大化地减少不必要的消耗。还有在节点内，相比于内存，磁盘IO对job完成时间的影响也是可观的。从最基本的要求来说，我们对Shuffle过程的期望可以有： 

• 完整地从map task端拉取数据到reduce 端。
• 在跨节点拉取数据时，尽可能地减少对带宽的不必要消耗。

减少磁盘IO对task执行的影响。
        OK，看到这里时，大家可以先停下来想想，如果是自己来设计这段Shuffle过程，那么你的设计目标是什么。我想能优化的地方主要在于减少拉取数据的量及尽量使用内存而不是磁盘。 
        我的分析是基于Hadoop0.21.0的源码，如果与你所认识的Shuffle过程有差别，不吝指出。我会以WordCount为例，并假设它有8个map task和3个reduce task。从上图看出，Shuffle过程横跨map与reduce两端，所以下面我也会分两部分来展开。 

        先看看map端的情况，如下图： 图二

 上图可能是某个map task的运行情况。拿它与官方图的左半边比较，会发现很多不一致。官方图没有清楚地说明partition， sort与combiner到底作用在哪个阶段。我画了这张图，希望让大家清晰地了解从map数据输入到map端所有数据准备好的全过程。 

        整个流程我分了四步。简单些可以这样说，每个map task都有一个内存缓冲区，存储着map的输出结果，当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘，当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并，生成最终的正式输出文件，然后等待reduce task来拉数据。 

        当然这里的每一步都可能包含着多个步骤与细节，下面我对细节来一一说明： 
1.  在map task执行时，它的输入数据来源于HDFS的block，当然在MapReduce概念中，map task只读取split。Split与block的对应关系可能是多对一，默认是一对一。在WordCount例子里，假设map的输入数据都是像“aaa”这样的字符串。 

2.在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“aaa”， value是数值1。因为当前map端只做加1的操作，在reduce task里才去合并结果集。前面我们知道这个job有3个reduce task，到底当前的“aaa”应该交由哪个reduce去做呢，是需要现在决定的。 

        MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。 

        在我们的例子中，“aaa”经过Partitioner后返回0，也就是这对值应当交由第一个reducer来处理。接下来，需要将数据写入内存缓冲区中，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。 

        整个内存缓冲区就是一个字节数组，它的字节索引及key/value存储结构我没有研究过。如果有朋友对它有研究，那么请大致描述下它的细节吧。 

3.  这个内存缓冲区是有大小限制的，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写，字面意思很直观。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size * spill percent = 100MB * 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写，互不影响。 

        当溢写线程启动后，需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为，这里的排序也是对序列化的字节做的排序。 

        在这里我们可以想想，因为map task的输出是需要发送到不同的reduce端去，而内存缓冲区没有对将发送到相同reduce端的数据做合并，那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于，如果有很多个key/value对需要发送到某个reduce端去，那么需要将这些key/value值拼接到一块，减少与partition相关的索引记录。 

        在针对每个reduce端而合并数据时，有些数据可能像这样：“aaa”/1， “aaa”/1。对于WordCount例子，就是简单地统计单词出现的次数，如果在同一个map task的结果中有很多个像“aaa”一样出现多次的key，我们就应该把它们的值合并到一块，这个过程叫reduce也叫combine。但MapReduce的术语中，reduce只指reduce端执行从多个map task取数据做计算的过程。除reduce外，非正式地合并数据只能算做combine了。其实大家知道的，MapReduce中将Combiner等同于Reducer。 

        如果client设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。 

4. 每次溢写会在磁盘上生成一个溢写文件，如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个溢写文件存在。当map task真正完成时，内存缓冲区中的数据也全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(如果map的输出结果很少，当map执行完成时，只会产生一个溢写文件)，因为最终的文件只有一个，所以需要将这些溢写文件归并到一起，这个过程就叫做Merge。Merge是怎样的？如前面的例子，“aaa”从某个map task读取过来时值是5，从另外一个map 读取时值是8，因为它们有相同的key，所以得merge成group。什么是group。对于“aaa”就是像这样的：{“aaa”, [5, 8, 2, …]}，数组中的值就是从不同溢写文件中读取出来的，然后再把这些值加起来。请注意，因为merge是将多个溢写文件合并到一个文件，所以可能也有相同的key存在，在这个过程中如果client设置过Combiner，也会使用Combiner来合并相同的key。 

        至此，map端的所有工作都已结束，最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息，如果reduce task得到通知，获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程开始启动。 

        

 

 

简单地说，reduce task在执行之前的工作就是不断地拉取当前job里每个map task的最终结果，然后对从不同地方拉取过来的数据不断地做merge，也最终形成一个文件作为reduce task的输入文件。见下图： 图三

 如map 端的细节图，Shuffle在reduce端的过程也能用图上标明的三点来概括。当前reduce copy数据的前提是它要从JobTracker获得有哪些map task已执行结束，这段过程不表，有兴趣的朋友可以关注下。Reducer真正运行之前，所有的时间都是在拉取数据，做merge，且不断重复地在做。如前面的方式一样，下面我也分段地描述reduce 端的Shuffle细节： 
1. Copy过程，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。因为map task早已结束，这些文件就归TaskTracker管理在本地磁盘中。 

2.  Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是，merge有三种形式：1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用，让人比较困惑，是吧。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。 

3. Reducer的输入文件。不断地merge后，最后会生成一个“最终文件”。为什么加引号？因为这个文件可能存在于磁盘上，也可能存在于内存中。对我们来说，当然希望它存放于内存中，直接作为Reducer的输入，但默认情况下，这个文件是存放于磁盘中的。当Reducer的输入文件已定，整个Shuffle才最终结束。然后就是Reducer执行，把结果放到HDFS上。 

 

 

 

 

 

 

 

 

 

 

 

 

 

HIVE----SQL解析引擎/数据仓库

 

Hive操作

Hive数据模型

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

Hbase——构建在HDFS上分布式面向列的存储系统

1.1.什么是hbase

HBASE是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，利用HBASE技术可在廉价PC Server上搭建起大规模结构化存储集群，处理由成千上万的行和列所组成的大型数据。

HBASE的目标是存储并处理大型的数据，更具体来说是仅需使用普通的硬件配置，就能够HBASE利用Hadoop HDFS作为其文件存储系统；HBASE利用Hadoop MapReduce来处理HBASE中的海量数据；HBASE利用Zookeeper作为协同服务。

 

1.2.与传统数据库的对比

1、传统数据库遇到的问题：

1）数据量很大的时候无法存储

2）没有很好的备份机制

3）数据达到一定数量开始缓慢，很大的话基本无法支撑

 2、HBASE优势：

1）线性扩展，随着数据量增多可以通过节点扩展进行支撑

2）数据存储在hdfs上，备份机制健全

3）通过zookeeper协调查找数据，访问速度块。

 

1.3.hbase集群中的角色

1、一个或者多个主节点，Hmaster

2、多个从节点，HregionServer

 

1.4.HBase中的表一般有这样的特点：

1 大：一个表可以有上亿行，上百万列

2 面向列:面向列(族)的存储和权限控制，列(族)独立检索。

3 稀疏:对于为空(null)的列，并不占用存储空间，因此，表可以设计的非常稀疏。

非关系型数据库

 

 

 

 

 

 

 

 

hbase数据模型  Row Key   Columns Family  Cell   Time Stamp

Row Key

与nosql数据库们一样,row key是用来检索记录的主键。访问HBASE table中的行，只有三种方式：

1.通过单个row key访问

2.通过row key的range（正则）

3.全表扫描

Row key行键 (Row key)可以是任意字符串(最大长度 是 64KB，实际应用中长度一般为 10-100bytes)，在HBASE内部，row key保存为字节数组。存储时，数据按照Row key的字典序(byte order)排序存储。

设计key时，要充分排序存储这个特性，将经常一起读取的行存储放到一起。(位置相关性)

 

Columns Family

列簇 ：HBASE表中的每个列，都归属于某个列族。列族是表的schema的一部 分(而列不是)，必须在使用表之前定义。列名都以列族作为前缀。例如 courses:history，courses:math都属于courses 这个列族。

 

Cell

由{row key, columnFamily, version} 唯一确定的单元。cell中 的数据是没有类型的，全部是字节码形式存贮。

关键字：无类型、字节码

 

Time Stamp

HBASE 中通过rowkey和columns确定的为一个存贮单元称为cell。

每个cell都保存着同一份数据的多个版本。

版本通过时间戳来索引。时间戳的类型是 64位整型。

时间戳可以由HBASE(在数据写入时自动 )赋值，此时时间戳是精确到毫秒的当前系统时间。

时间戳也可以由客户显式赋值。如果应用程序要避免数据版本冲突，就必须自己生成具有唯一性的时间戳。

每个 cell中，不同版本的数据按照时间倒序排序，即最新的数据排在最前面。

为了避免数据存在过多版本造成的的管理 (包括存贮和索引)负担，HBASE提供 了两种数据版本回收方式。

一是保存数据的最后n个版本，二是保存最近一段 时间内的版本（比如最近七天）。用户可以针对每个列族进行设置。

Hbase命令

hbase shell

创建表                       create '表名', '列族名1','列族名2','列族名N'

查看所有表                   list

描述表                       describe  ‘表名’

判断表存在                   exists  '表名'

判断是否禁用启用表           is_enabled '表名'   is_disabled ‘表名’

添加记录                     put  ‘表名’, ‘rowKey’, ‘列族 : 列‘  ,  '值'

查看记录rowkey下的所有数据   get  '表名' , 'rowKey'

查看表中的记录总数           count  '表名'

获取某个列族               get '表名','rowkey','列族'

获取某个列族的某个列       get '表名','rowkey','列族：列’

删除记录                   delete  ‘表名’ ,‘行名’ , ‘列族：列'

删除整行                   deleteall '表名','rowkey'

删除一张表                   先要屏蔽该表，才能对该表进行删除

                              第一步 disable ‘表名’ ，第二步  drop '表名'

清空表                       truncate '表名'

查看所有记录               scan "表名"  

查看某个表某个列中所有数据   scan "表名" , {COLUMNS=>'列族名:列名'}

更新记录                 就是重写一遍，进行覆盖，hbase没有修改，都是追加

 

hbase过滤器

FilterList

FilterList 代表一个过滤器列表，可以添加多个过滤器进行查询，多个过滤器之间的关系有：

与关系（符合所有）：FilterList.Operator.MUST_PASS_ALL  

或关系（符合任一）：FilterList.Operator.MUST_PASS_ONE

 

过滤器的种类：

列植过滤器           SingleColumnValueFilter           过滤列植的相等、不等、范围等          

列名前缀过滤器       ColumnPrefixFilter                过滤指定前缀的列名

多个列名前缀过滤器   MultipleColumnPrefixFilter        过滤多个指定前缀的列名

rowKey过滤器         RowFilter                         通过正则，过滤rowKey值。

 

Hbase原理： Hbase体系原理图

 写流

1、client向hregionserver发送写请求。

2、hregionserver将数据写到hlog（write ahead log）。为了数据的持久化和恢复。

3、hregionserver将数据写到内存（memstore）

4、反馈client写成功。

 

数据flush过程

1、当memstore数据达到阈值（默认是64M），将数据刷到硬盘，将内存中的数据删除，同时删除Hlog中的历史数据。

2、并将数据存储到hdfs中。

3、在hlog中做标记点。

 

数据合并过程

1、当数据块达到4块，hmaster将数据块加载到本地，进行合并

2、当合并的数据超过256M，进行拆分，将拆分后的region分配给不同的hregionserver管理

3、当hregionserver 宕机后，将hregionserver上的hlog拆分，然后分配给不同的hregionserver加载，修改.META.

4、注意：hlog会同步到hdfs

 

hbase的读流程

1、通过zookeeper和-ROOT- .META.表定位hregionserver。

2、数据从内存和硬盘合并后返回给client

3、数据块会缓存

 

hmaster的职责

1、管理用户对Table的增、删、改、查操作；

2、记录region在哪台Hregion server上

3、在Region Split后，负责新Region的分配；

4、新机器加入时，管理HRegion Server的负载均衡，调整Region分布

5、在HRegion Server宕机后，负责失效HRegion Server 上的Regions迁移。

 

hregionserver的职责

HRegion Server主要负责响应用户I/O请求，向HDFS文件系统中读写数据，是HBASE中最核心的模块。

HRegion Server管理了很多table的分区，也就是region。

 

client职责

Client

HBASE Client使用HBASE的RPC机制与HMaster和RegionServer进行通信

管理类操作：Client与HMaster进行RPC；

数据读写类操作：Client与HRegionServer进行RPC。