HBase/Hadoop学习笔记 (转)

HBase/Hadoop学习笔记

学习目标：

至少掌握五点：

1.    深入理解HTable，掌握如何结合业务涉及高性能的HTable。

2.    掌握与HBase的交互，通过HBase Shell命令及Java API进行数据的增删改查。

3.    掌握如何用MapReduce分析HBase里的数据

4.    掌握如何测试HBase MapReduce。

HBase简介：

         HBase在产品中还包含了Jetty，在HBase启动时采用嵌入式的方式来启动Jetty，因此可以通过web界面对HBase进行管理和查看当前运行的一些状态，非常轻巧。

简单来说，你在HBase中的表创建的可以看做是一张很大的表，而这个表的属性可以根据需求去动态增加，在HBase中没有表与表之间关联查询。

列存储

HStore存储是HBase存储的核心了，其中由两部分组成，一部分是MemStore，一部分是StoreFiles。MemStore是 Sorted Memory Buffer，用户写入的数据首先会放入MemStore，当MemStore满了以后会Flush成一个StoreFile（底层实现是HFile）， 当StoreFile文件数量增长到一定阈值，会触发Compact合并操作，将多个StoreFiles合并成一个StoreFile，合并过程中会进 行版本合并和数据删除，因此可以看出HBase其实只有增加数据，所有的更新和删除操作都是在后续的compact过程中进行的，这使得用户的写操作只要 进入内存中就可以立即返回，保证了HBase I/O的高性能。当StoreFilesCompact后，会逐步形成越来越大的StoreFile，当单个StoreFile大小超过一定阈值后，会触 发Split操作，同时把当前Region Split成2个Region，父Region会下线，新Split出的2个孩子Region会被HMaster分配到相应的HRegionServer 上，使得原先1个Region的压力得以分流到2个Region上。

HFile里面的每个KeyValue对就是一个简单的byte数组。

 HRegion是Hbase中分布式存储和负载均衡的最小单元。最小单元就表示不同的Hregion可以分布在不同的HRegion server上。但一个Hregion是不会拆分到多个server上的。

HFile格式：

一、HBase基本概念：

2个主要概念：

a) Rowkey: Hbase 中的记录是按照rowkey来排序的；

b) Column family：(列族)是在系统启动之前预先定义好的；

c) Hbase优缺点：

1.不支持条件查询以及orderby等查询；

2.列可以动态增加，列为空则不存储数据，节省存储空间；

3.会自动切分数据；4.可以提供高并发读写操作的支持；

访问方式： 访问hbasetable中的行，只有三种方式：

1 通过单个row key访问

2 通过row key的range

3 全表扫描

Row key：行键 (Row key)可以是任意字符串(最大长度是 64KB，实际应用中长度一般为 10-100bytes)，在hbase内部，row key保存为字节数组。

存储：  HBase以表的形式存储数据。表有行和列组成，存储时，数据按照Row key的字典序(byte order)排序存储。设计key时，要充分排序存储这个特性，将经常一起读取的行存储放到一起。(位置相关性) 

存储类型：  TableName 是字符串

RowKey 和 ColumnName 是二进制值（Java 类型 byte[]）

Timestamp 是一个 64 位整数（Java 类型 long）

value 是一个字节数组（Java类型 byte[]）

存储结构：即HTable按Row key自动排序，每个Row包含任意数量个Columns，Columns之间按Columnkey自动排序，每个Column包含任意数量个Values。理解该存储结构将有助于查询结果的迭代。

        (RowKey,List(SortedMap(column,List(value,TimeStamp))))

列簇：hbase表中的每个列，都归属与某个列族。列名都以列族作为前缀。

         HBase中的列可以动态新增。

存储单元：HBase中通过row和columns确定的为一个存贮单元称为cell。

            每个cell都保存着同一份数据的多个版本。版本通过时间戳来索引。

HBase为null的Column不会被存储，这样既节省了空间又提高了读性能

cell中的数据是没有类型的，全部是字节码形式存贮

两种数据版本回收方式：一是保存数据的最后n个版本

二是保存最近一段时间内的版本（比如最近七天）

用户可以针对每个列族进行设置值value：每个值由4个键唯一索引

tableName+RowKey+ColumnKey+Timestamp=>value

二、搭建HBase环境：

http://hbase.apache.org/book/quickstart.html和http://hbase.apache.org/book/notsoquick.html。 如果你在windows环境下配置cygwin及ssh遇到问题可以参考http://qa.taobao.com/?p=10633

1.    创建一个Maven工程。

mvn archetype:generate-DgroupId=com.alibaba.webx -DartifactId=tutorial1 -Dversion=1.0-SNAPSHOT -Dpackage=com.alibaba.webx.tutorial1-DarchetypeArtifactId=archetype-webx-quickstart-DarchetypeGroupId=com.alibaba.citrus.sample-DarchetypeVersion=1.0-SNAPSHOT -DinteractiveMode=false

 cmd进入刚才建立的项目，运行：mvn jetty:run   在浏览器中打开：localhost:8081就可以看到我们新建的webx项目了。具体里面是怎么运行的，可以查看webx用户手册。

2.    加入Hadoop、HBase依赖：

3.    在src/test/resources目录下新建文件conf/hbase-site.xml

文件具体配置为：

1. <configuration>  
2.   <property>  
3.      <name>hbase.zookeeper.property.clientPort</name>  
4. 2222</value>  
5.    </property>  
6.    <property>  
7.      <name>hbase.zookeeper.quorum</name>  
8.       <value>xxx.xxx.xxx.xxx.xxx</value>  
9.    </property>  
10. "true">  
11. for.version</name>  
12. 0.90.2</value>  
13.   </property>  
14.  </configuration>

配置你工程要使用的zookeeper客户端端口号和zookeeper的地址，这个地址可以向开发索要。

4.    写测试文件：

1. public class
2. private static Configuration conf = null;  
3. static
4.             conf = HBaseConfiguration.create();  
5. "conf/hbase-site.xml");  
6.       }}

       如果不报错，表示链接已经通过，接下来就可以创建表以及对表的增删改查了。

三、基础知识：

1.     通过HBase shell 与HBase交互：

              进入控制台：bin/hbase shell

              创建表：create ‘表名’,’列簇名’,’列簇名’

              增加记录：put ‘表名’,’Row Key’,’列簇名：列名’,’value’

              查询：get ‘表名’,’Row Key’

              删除：delete‘表名’,’Row Key’,’列簇名：列名’  （只能删除一列）

                      delete‘表名’,’Row Key’                    （删除RowKey的所有列）

              删除表：>disable ‘表名’

                        >drop ‘表名’

2.     通过Java 的API与HBase交互：

              步骤一：

                     创建一个Maven工程加入依赖：

   
1.                   <dependency>  
2. <groupId>org.apache.hbase</groupId>  
3. <artifactId>hbase</artifactId>  
4. <version>0.90.2</version>  
5. </dependency>

如果你的Maven库里还没有hbase，还需要配置下repository:

1.        <repositories>  
2. <repository>  
3. <id>cloudera</id>  
4. <url>https://repository.cloudera.com/content/groups/public</url>
5. </repository>  
6. </repositories>

                     步骤二：

  确保HBase环境已启动且能连接到，将HBase环境的hbase-site.xml文件拷贝到上述工程的src/test/resources目录

                     加载配置->创建表->增加记录->根据RowKey查询->遍历查询与迭代->删除记录->删除表

                  具体操作可以参考：http://qa.taobao.com/?p=13894

四、深入理解HBase：

思考：HBase服务器内部由那些主要部件构成？

HBase的内部工作原理是什么？

1.    HBase的工作原理：

                  首先HBase Client端会连接Zookeeper Qurom(从下面的代码也能看出来，例 如：HBASE_CONFIG.set("hbase.zookeeper.quorum","192.168.50.216") )。通过Zookeeper组件Client能获知哪个Server管理-ROOT-Region。那么Client就去访问管理-ROOT-的 Server，在META中记录了HBase中所有表信息，(你可以使用 scan '.META.' 命令列出你创建的所有表的详细信息),从而获取Region分布的信息。一旦Client获取了这一行的位置信息，比如这一行属于哪个 Region，Client将会缓存这个信息并直接访问HRegionServer。久而久之Client缓存的信息渐渐增多，即使不访问.META.表 也能知道去访问哪个HRegionServer。HBase中包含两种基本类型的文件，一种用于存储WAL的log，另一种用于存储具体的数据，这些数据 都通过DFS Client和分布式的文件系统HDFS进行交互实现存储。

2.    Client访问数据过程：

Client访问用户数据之前需要首先访问zookeeper，然后访问-ROOT-表，接着访问.META.表，最后才能找到用户数据的位置去访问，中间需要多次网络操作，不过client端会做cache缓存。

-ROOT-表、.META都是存放在哪里？？

client访问hbase上数据的过程并不需要master参与（寻址访问zookeeper和region server，数据读写访问region server），master仅仅维护者table和region的元数据信息，负载很低。

3.    在HBase上进行MapReduce操作：

4.    HBase系统架构：

　　        HBase Client使用HBase的RPC机制与HMaster和HRegionServer进行通信，对于管理类操作，Client与HMaster进行RPC；对于数据读写类操作，Client与HRegionServer进行RPC

5.    Zookeeper：

                  Zookeeper简单说就是协调和服务于分布式应用程序的服务。

　　        Zookeeper Quorum中除了存储了-ROOT-表的地址和HMaster的地址，HRegionServer也会把自己以Ephemeral方式注册到 Zookeeper中，使得HMaster可以随时感知到各个HRegionServer的健康状态。此外，Zookeeper也避免了HMaster的 单点问题。

1 保证任何时候，集群中只有一个master

2存贮所有Region的寻址入口。

3 实时监控RegionServer的状态，将Region server的上线和下线信息实时通知给Master

4 存储Hbase的schema,包括有哪些table，每个table有哪些column family

Zookeeper到底为我们干了什么？

1.    集中配置：可以APP1的配置配置到/APP1 znode下的所有机器。

2.    集群管理：同步：维护活机列表(让集群所有机器得到实时更新)，

组服务：从集群中选择Master。

3.    …..

6.    HMaster：

　　HMaster没有单点问题，HBase中可以启动多个HMaster，通过Zookeeper的MasterElection机制保证总有一个Master运行，HMaster在功能上主要负责Table和Region的管理工作：

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

　　2. 管理HRegionServer的负载均衡，调整Region分布

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

　　4. 在HRegionServer停机后，负责失效HRegionServer 上的Regions迁移

7.    HRegionServer：

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

　　HRegionServer内部管理了一系列HRegion对象，每个HRegion对应了Table中的一个Region，HRegion中 由多个HStore组成。每个HStore对应了Table中的一个ColumnFamily的存储，可以看出每个Column Family其实就是一个集中的存储单元，因此最好将具备共同IO特性的column放在一个ColumnFamily中，这样最高效。

8.    Hadoop+HBase+Zookeeper三者关系：

1.经过Map、Reduce运算后产生的结果看上去是被写入到HBase了，但是其实HBase中HLog和StoreFile中的文件在进行flush to disk操作时，这两个文件存储到了HDFS的DataNode中，HDFS才是永久存储。

2.ZooKeeper跟HadoopCore、HBase有什么关系呢？ZooKeeper都提供了哪些服务呢？主要有：管理Hadoop集群中 的NameNode，HBase中HBaseMaster的选举，Servers之间状态同步等。具体一点，细一点说，单只HBase中 ZooKeeper实例负责的工作就有：存储HBase的Schema，实时监控HRegionServer,存储所有Region的寻址入口，当然还有 最常见的功能就是保证HBase集群中只有一个Master。

Hadoop、ZooKeeper和HBase之间应该按照顺序启动和关闭：启动Hadoop—>启动ZooKeeper集群—>启动HBase—>停止HBase—>停止ZooKeeper集群—>停止Hadoop。

参考：http://www.tbdata.org/archives/1509

五：理解Hadoop：

Hadoop学习网址：http://book.51cto.com/art/201106/269616.htm

1.    MapReduce

(1)   MapReduce基础：
一个简单的MapReduce程序需要三样东西
1. 实现Mapper，处理输入的对，输出中间结果
2. 实现Reduce，对中间结果进行运算，输出最终结果
3. 在main方法里定义运行作业，定义一个job，在这里控制job如何运行等。

Mapper接口：是一个泛型，有4个形式的参数类型，分别指定map函数的输入键，输入值，输出键，输出值。

数据类型：Hadoop规定了自己的一套可用于网络序列优化的基本类型，而不是使用内置的java类型，这些都在 org.apache.hadoop.io包中定义，上面使用的Text类型相当于java的String类型，IntWritable类型相当于 java的Integer类型。

          Maper  和Reducer 可以理解为分久必合，合久必分！

                            Maper是将任务切分成很多个小任务，分配给不同的工作者去完成

                          Reducer是将哪些工作者做完的工作结果收集起来加以整理汇总成最后结果。

总结：job的配置有着复杂的属性参数，如文件分割策略、排序策略、map输出内存缓冲区的大小、工作线程数量等，深入理解掌握这些参数才能使自己的MapReduce程序在集群环境中运行的最优。

(2)深入理解MapReduce：

        （1）在map进行之前，需要对输入文件在客户端先进行“分片”，然后将分片信息上传到HDFS。

        （2）分片上传结束后，jobtracker拿到分片信息，来分配map,reduct task；map对每条记录的输出以<key,value> 的形式输出。

        （3）如果定义了combiner，则在本地会对map处理的结果进行处理：对相同key的聚合，对key的排序，value的迭代。combiner完成类似于本地reduce的功能。

        （4）在进入reduce阶段之前，系统会完成一些列操作（merge，sort）：将list中key相同的数据进行合并、排序，最后形成<k1`,list<v1`>>的数据；

             然后发往一个reduce

        （5）进入一个reduce，相同的key的map输出会到达同一个reduce，reduce对key相同的多个value进行“reduce操作”；

        > 没有combiner的处理过程：

        > 添加combiner的处理过程：

         ？为什么我买的map函数和reduce函数一般使用静态类？

           答：task内部可以共享静态类属性，每个task可能会多次调用map或reduce函数，但每个key只对应某个节点上的某个task的reduce函数的一次执行。

               多个task之间不能共享静态类属性，即使是在同一台机器上，因为是以进程的方式在运行。

Map类：(继承TableMapper或者Mapper)

           Map原理：

                在map阶段，使用job.setInputFormatClass定义的InputFormat将输入的数据集分割成小数据块splites，同时 InputFormat提供一个RecordReder的实现。本例子中使用的是                        TextInputFormat，他提供的RecordReder会将文本的一行的行号作为key，这一行的文本作为value。这就是自定义Map的输 入是<LongWritable,Text>的原因。然后调用自定义Map的map方法，将一个个<LongWritable, Text>对输入给Map的map方法。注意输出应该符合自定义Map中定义的输出<IntPair, IntWritable>。最终是生成一个List<IntPair,IntWritable>。在map阶段的最后，会先调用 job.setPartitionerClass对这个List进行分区，每个分区映射到一个reducer。每个分区内又调用 job.setSortComparatorClass设置的key比较函数类排序。可以看到，这本身就是一个二次排序。如果没有通过 job.setSortComparatorClass设置key比较函数类，则使用key的实现的compareTo方法。在第一个例子中，使用了 IntPair实现的compareTo方法，而在下一个例子中，专门定义了key比较函数类。

                 Q: map的结果发给那个reduce？谁来管理这一切？

                 A:     Partitioner用于划分键值空间（key space）。

Hash函数。分区的数目与一个作业的reduce任务的数目是一样的。因此，它控制将中间过程的key（也就是这条记录）应该发送给m个reduce任务中的哪一个来进行reduce操作。

Reduce类:(继承TableReducer或者Reducer)

           Reduce的原理：在reduce阶段，reducer接收到所有映射到这个reducer的map输出后，也是会调用 job.setSortComparatorClass设置的key比较函数类对所有数据对排序。然后开始构造一个key对应的value迭代器。这时就 要用到分组，使用jobjob.setGroupingComparatorClass设置的分组函数类。只要这个比较器比较的两个key相同，他们就属 于同一个组，它们的value放在一个value迭代器，而这个迭代器的key使用属于同一个组的所有key的第一个key。最后就是进入Reducer 的reduce方法，reduce方法的输入是所有的（key和它的value迭代器）。同样注意输入与输出的类型必须与自定义的Reducer中声明的 一致。

                         reduce的输出是没有排序的。

           Q: Reduce的数目应该设置多少？

           A:Reduce的数目建议是0.95或1.75乘以 ( *mapred.tasktracker.reduce.tasks.maximum)。用0.95，所有reduce可以在maps一完成时就立刻启 动，开始传输map的输出结                            果。用1.75，速度快的节点可以在完成第一轮reduce任务后，可以开始第二轮，这样可以得到比较好的负载均衡的效果。上述比例因子比整体数目稍小一 些是为了给框                            架中的推测性任务（speculative-tasks） 或失败的任务预留一些reduce的资源。

           Q: Reduce的三个阶段都干了什么？

           A: Reducer有3个主要阶段：shuffle、sort和reduce。

               Shuffle ：Reducer的输入就是Mapper已经排好序的输出。在这个阶段，框架通过HTTP为每个Reducer获得所有Mapper输出中与之相关的分块。（其实就是copy的过程）

               Sort ：这个阶段，框架将按照key的值对Reducer的输入进行分组 （因为不同mapper的输出中可能会有相同的key，combain保证了同一台机器相同key的合并，但是不同机器也可能有相同的key）。

                map的输出是一边被取回一边被合并的。

Job 的配置：

           (1). 使用job.setInputFormatClass定义的InputFormat将输入的数据集分割成小数据块splites,HadoopMap/Reduce框架为每一个Split产生一个map任务.

             Map的数目通常是由输入数据的大小决定的，一般就是所有输入文件的总块（block）数。如果你输入10TB的数据，每个块（block）的大小是 128MB，你将需要大约82,000个map来完成任务，除非使用setNumMapTasks(int)将这个数值设置得更高。

           (2).如果需要中间过程对key的分组规则和reduce前对key的分组规则不同，那么可以通过 JobConf.setOutputValueGroupingComparator(Class)来指定一个Comparator。再加上 JobConf.setOutputKeyComparatorClass(Class)可用于控制中间过程的key如何被分组，所以结合两者可以实现按 值的二次排序

           (3).一些作业的参数可以被直截了当地进行设置（例如： setNumReduceTasks(int)），而另一些参数则与框架或者作业的其他参数之间微妙地相互影响，并且设置起来比较复杂（例 如：              setNumMapTasks(int)）

           (4).Mapper和Reducer的实现可以利用Reporter 来报告进度，或者仅是表明自己运行正常。我们从界面上看到的图形就是利用Reporter来进行进度的展示。

(2)    MapReduce基本编程：

Ø  创建一个Maven工程

Ø  加入hadoop依赖

Ø  编写Map类

Ø  编写reduce类

Ø  定义job

(3)    进行Mapreduce测试：

l   用MRUnit做单元测试：

Ø  加入mrunit依赖

Ø  单独测试Map

Ø  单独测试Reduce

Ø  测试MapReduce

参考：叶渡：Hadoop学习笔记_yedu.pdf

疑惑：1. 使用MRUnit，测试代码在.run下通过，在.runTest()失败，原因是什么？两者有什么区别？

l   运行MapReduce  Job进行集成测试

流程：预设置(准备输入文件、启动hadoop进程等)->运行作业->输出结果跟预期结果的对比->报告导致失败的原因

l   精简HBaseMapReduce测试：

使用Hadoop/HBaseMini Cluster (iTest-hadoop)

参考文档：http://qa.taobao.com/?p=13939

（不安装Hadoop、HBase环境，只要有JDK搞定MapReduce的Job测试）

2.    本地搭建单机版hadoop环境（win）：

            Hadoop主要是在Linux 平台下运行的，如果想在 Windows 平台下运行，你需要安装 Cygwin 才能运行， Hadoop 脚本。

         按照“在Windows上安装Hadoop教程.pdf“执行完成。

         安装还可以参考文档：

         按照“在Windows上安装Hadoop教程.pdf“的说明进行到最后一步时，在启动./start-all.sh之前，需要 格式化一个新的分布式文件，./hadoopnamenode –format .这样就会启动JobTracker.

         浏览NameNode 和JobTracker 的网络接口，他们的地址默认为：

              NameNode – http://localhost:50070/

              JobTracker  – http://localhost:50030/

         Node数为0，如何配置NameNode 和Datanode？？

Ø  运行hadoop自带jar文件：

        运行hadoop自带的jar文件，理解MapReduce的过程：hadoop-0.20.2-examples.jar

        跑通自己第一个Job程序：

              首先开启hadoop服务： ./start-all.sh

1.    根据文档示例编写wordCout程序。

2.    将编写的代码打包成HadoopTest.jar放到本地某一个目录下， (打包的时候要选择mainclass)

或者直接运行hadoop自带文件中的示例jar包(hadoop-0.20.2-examples.jar)。

3.    将要分析的数据传到hdfs上去

在dfs上创建测试输入目录：./hadoop dfs –mkdir test-in

然后将本地文件copy到test-in中: ./hadoop dfs –copyFromLocal [本地文件目录] test-in

验证文件是否复制成功： ./hadoop dfs –ls test-in

 注：这里的test-in其实是HDFS路径下的目录，七绝对路径为

“http://localhost:50070/user/XXXXX/test-in”

      4. 开始执行

./bin/hadoopjar hadoop-0.20.2-examples.jar wordCount test-in test-out

当遇到文件已存在异常的时候，只要将test-out改一个名字即可。

5.    遇到问题：

抛出文件不存在的异常，原因是找不到tmp目录。开如图上的目录看到并不存在那样的目录结构，说明根本就没有创建相应的目录结构，可能是连tmp都没有找到，所以查找配置文件发现，conf下的mapred-site.xml中中默认配置是

./tmp,所以修改成自己的相应目录就可以了。

6.    运行成功：

Ø  运行自己编写的文件：

(1).逻辑性代码：

1.    编写自己的mapper函数：继承Mapper基类，实现map方法

2.    编写自己的reducer函数：继承reducer基类，实现reduce方法

3.    编写自己的主函数：创建job，配置map、combiner、reducer类型，设置输入输出路径，                  设置输出键/值格式，提交任务

(2).驱动性代码：驱动类来注册业务的class为一个可标示的命令，让hadoop jar可以执行。

       如：

(3). 最后一步：将自己的项目导出成jar格式，注意：在选择main class时，是选择我们创建的驱动类，而不是逻辑主类。

3. 一个Job的请求过程：

用户通过界面提交一个Job，服务器把Job请求发送给gateway，gateway接收请求后按照一定的逻辑拼装成MR需要的请求文件。

Gateway：我把可以把gateway理解为跳板机，我们的机器不能直接访问集群，需要一个入口，这个入口就是Gateway。

思考：这个跳板机是单独拿出来的一台机器专门做Job的入口的呢，还是只是机群中的普通机器？

JobTracker：

TaskTracker：

我们提交一个JOB（一般通过JobClient，这个类有三种策略来提交一个JOB，1、job完成后才返回状态2、job提交后，返回一个持有状态的Handler，3、提交job，但是不返回状态）

首先会从JobTracker（hadoop中运用了master/slaver机制，他是master服务，那么slaver在这里就是tasktracker）中的得到一个job的definition Id, 其实这个id也就是JobTracker管理job的个数

jobClient会从Configuration找到hadoop系统目录（"mapred.system.dir",默认值"/tmp /hadoop/mapred/system"）在这里jobClient做了一件比较重要的事情，他把input的数据做split操作（相当于将大数 据量切分成若干块，具体切分成多大，这个通过一个公式来计算的：FileInputFormat的策略max(minimumSize,min(maximumSize, blockSize))  其中minSize表示一个map切分的最小容量，maxSize即最大容量，blockSize表示HDFS中的block容量）[1]，从而决定了 Map的个数（其实就是MapTask的数量）。jobClient还将一部分资源文件放到jobtracker的FS中（jar、file、 archives、split[2]）

正式提交

JT(jobtracker)会根据这个job创建一个JobInProgress对象，这个对象记录着这个job所有信息。最后JT会将这个 job注册到JobInProgressListener中（以下简称JIPL），让JIPL监听这些job。JIPL是在JT启动的时候启动的监听器 （由TaskScheduler注入，作用参照step 7）。一个EagerTaskInitializationListener：它是一个生命周期和JT一样的监听线程，主要功能就是初始化这个Job,并 且创建相应的TaskInProgress(TIP,包括M个MapTask,N个ReduceTask,2个CleanTask,2个 SetupTask)。另一个就是JobQueueJobInProgressListener，这个listener是处理job队列的，也就是job 提交的先后顺序跟它有关系，默认的是FIFO。

在初始化job的时候（其实是初始化MapTask）,会将之前的input数据split的信息回流回来，初始化maptask

到这里，JT初始化job工作完毕。

每台slaver机器启动的时候，都会启动一个tasktracker的线程，这个线程主要负责和JT去通信，也就是发送心跳（通过RPC通信协 议）。当发送心跳的时候，TT会将自己现有的状态（是否是刚刚启动、是否刚初始化，自己状态是否可以申请新的task，如果JT中没有这个TT的引用，那 么需要保存下来）JT首先会获取Setup和CleanUp的Task（默认每个job都会有两个setuptask和两个cleanup task），如果没有了以上两种类型的task，那么剩下的就是MapTask和ReduceTask此时，JT会去向TaskScheduler这个调 度类去申请Task。在hadoop中，默认的TaskScheduler是JobQueueTaskScheduler，他持有JPL的引用。当TT发 送一个心跳表明自己空闲需要执行Task时，这时候，JT会调用Schedule的assignTask方法去获得一个Task。（这里Hadoop找 MapTask的时候，首先node-local，然后rack-local，最后才是不同机架，具体怎么找，还未仔细看）

TT发送心跳后，JT返回给TT一个HeartbeatResponse对象的引用，这里面包含着需要执行Tasks的action数组（如果 action的类型LaunchTaskAction：执行一个新的Task, 如果CommitTaskAction：加入commitResponses列表，由Task在适当的时候提交给JobTracker），同时JT还会更 新TTS的内容。所以TT根据这两点，就可以很好的判断自己Task在JT那边的状态。 执行任务前先调用localizeTask()更新一下jobConf文件并写入到本地目录中。然后通过调用Task的createRunner()方法 创建TaskRunner对象并调用其start()方法，值得注意点是，TaskRunner会去新开一个JVM去执行Task(如果考虑掉Task开 销小而且多，可以将jvm reuse)。

两种启动Job方式：

A:Submit()

submit函数会把Job提交给对应的Cluster，然后不等待Job执行结束就立刻返回。同时会把Job实例的状态设置为 JobState.RUNNING，从而来表示Job正在进行中。然后在Job运行过程中，可以调用getJobState()来获取Job的运行状态。

B:waitForCompletion(boolean)

waitForCompletion函数会提交Job到对应的Cluster，并等待Job执行结束。函数的boolean参数表示是否打印Job执行的相关信息。返回的结果是一个boolean变量，用来标识Job的执行结果。

执行Job的内部流程：

1).Inputformat会从job的INPUT_DIR目录下读入待处理的文件，检查输入的有效性并将文件切分成InputSplit列表。 Job实例可以通过setInputFormatClass(Class<? extends InputFormat>)函数来设置所需的inputformat。

2).当Inputformat对输入文件分片后，会对每个分片构建一个MapperTask实例（MapTask(String, TaskAttemptID, int, TaskSplitIndex, int)）。其实整个Mapper类的调度过程，都是由MapperTask来实现的。MapperTask的 run(JobConf,TaskUmbilicalProtocol)方法实现了对于Mapper task调度的整个过程。

2.1)  RecordReader会解析InputSplit，生成对应的key/value pair。Inputformat中有一个除了用于分片的getSplits(JobContext)方法外，还有一个方法 createRecordReader(InputSplit,TaskAttemptContext)，该方法用于给每一个分片创建一个 RecordReader。重写这个方法，可以添加自己的RecordReader。

2.2)Mapper类会对属于一个InputSplit所有key/value pair调用一次map函数。关于Mapper类的作用，在Javadoc中描述如下：“Mapper maps input key/value pairs to a set ofintermediate key/value pairs”。 Job实例可以通过setMapperClass(Class<? extends Mapper>)函数来设置自己的Mapper类。

2.3)可以通过Job实例的setSortComparatorClass(Class<?extends RawComparator>)方法来为Mapper设定一个Comparator class，用来对Mapper的结果根据key进行排序。

2.4)可以通过Job实例的setPartitionerClass(Class<? extends Partitioner>)方法来为Mapper设定一个Partitioner Class，用来对Mapper的结果根据Reducer进行分片。

2.5)可以通过Job实例的setCombinerClass(Class<? extends Reducer>)方法为Mapper设定一个Combiner Class，用来在本地进行聚集操作，从而减少从Mapper到Reducer的数据传输量。

3).Mapper执行结束之后，ReducerTask类会被用来进行整个Reducer操作的调度

3.1)Shuffle类会被调用从而来获取在Mapper输出中属于本Reducer的分片，并将多个分片combine成一个。

3.2)Shuffle类会使用MergeManager根据Job实例的 setSortComparatorClass(Class<?extends RawComparator>)所设定的Comparator class对key/value pair进行排序

3.3)在shuffle操作执行结束之后，如果对于Reducer的input数据，有使用特殊分组的需求的话，可以通过Job实例的 setGroupingComparatorClass(Class<?extends RawComparator>)方法来实现定制的分组策略，否则，则使用 setSortComparatorClass(Class<?extends RawComparator>)的比较方式。

3.4)在分组后的结果中，针对每一个<key, (list of values)> pair 调用Reduce的reduce(K2, Iterator<V2>, OutputCollector<K3, V3>,Reporter)方法。可以通过Job实例的setReducerClass(Class<?extends Reducer>)方法类设置相应的Reduce实现。

4).Reduce的结果将由OutputCollector.collect(WritableComparable, Writable)写入文件系统

3.    启动Hadoop过程：

(1)  启动NameNode:

启动NameNode节点；

初始化操作(如在name目录下创建文件)；

记录HDFS状态(如安全模式状态)；

本机FS注册，启动HDFS容器，并初始化；

(2)  启动DataNode:

(3)  启动SecondaryNameNode:

(4)  启动JobTracker:

(5)  启动TaskTracker:

4.    运行Map，Reduce过程：

1.    在分布式环境中客户端创建任务并提交。

2.    InputFormat做Map前的预处理，主要负责以下工作：

1.    验证输入的格式是否符合JobConfig的输入定义，这个在实现Map和构建Conf的时候就会知道，不定义可以是Writable的任意子类。

2.    将input的文件切分为逻辑上的输入InputSplit，其实这就是在上面提到的在分布式文件系统中blocksize是有大小限制的，因此大文件会被划分为多个block。

3.    通过RecordReader来再次处理inputsplit为一组records，输出给Map。 （inputsplit只是逻辑切分的第一步，但是如何根据文件中的信息来切分还需要RecordReader来实现，例如最简单的默认方式就是回车换行 的切分）

3.    RecordReader处理后的结果作为Map的输入，Map执行定义的Map逻辑，输出处理后的key和value对应到临时中间文件。

4.    Combiner可选择配置，主要作用是在每一个Map执行完分析以后，在本地优先作Reduce的工作，减少在Reduce过程中的数据传输量。

5.    Partitioner可选择配置，主要作用是在多个Reduce的情况下，指定Map的结果由某一个Reduce处理，每一个Reduce都会有单独的输出文件。（后面的代码实例中有介绍使用场景）

6.    Reduce执行具体的业务逻辑，并且将处理结果输出给OutputFormat。

7.    OutputFormat的职责是，验证输出目录是否已经存在，同时验证输出结果类型是否如Config中配置，最后输出Reduce汇总后的结果。

5.    MapReduce 中如何处理HBase中的数据？如何读取HBase数据给Map？如何将结果存储到HBase中？

Mapper类：包括一个内部类(Context)和四个方法(setup,map,cleanup,run)；

         setup,cleanup用于管理Mapper生命周期中的资源。setup -> map -> cleanup，

 run方法执行了这个过程；

        map方法用于对一次输入的key/value对进行map动作，对应HBase操作也就是一行的处理；

job的配置：

5.1TableInputFormat完成了什么功能？

        (1)通过设置conf.set(TableInputFormat.INPUT_TABLE,"udc_sell");设定HBase的输入表；

                设置conf.set(TableInputFormat.SCAN,TableMRUtil.convertScanToString(scan));设定对HBase输入表的scan方式；

        (2)通过TableInputFormat.setConf(Configration conf)方法初始化scan对象；

            scan对象是从job中设置的对象，以字符串的形式传给TableInputFormat，在TableInputFormat内部将scan字符创转换为scan对象

      * TableMapReduceUtily有两个方法：convertScanToString和convertStringToScan作用？

Base64字符串  和将Base64字符串还原为scan实例；

             Q:为什么不直接穿Scan对象而是费尽周折地转换来转换去呢？

             A:

         (3)TableInputFormat继承了TableInputFormatBase实现了InputFormat抽象类的两个抽象方法：

            getSplits()和createRecordReader()方法：

l  getSplits()断定输入对象的切分原则：对于TableInputFormatBase，会遍历 HBase相应表的所有HRegion，每一个HRegion都会被分成一个split，所以切分的块数是与表中HRegion的数目是相同的； InputSplitsplit =newTableSplit(table.getTableName(),splitStart, splitStop, regionLocation);在split中只会记载HRegion的其实rowkey和终止rowkey，具体的去读取这篇区域的数据是createRecordReader()实现的。

              计算出来的每一个分块都将被作为一个map Task的输入；

                 Q:但是分出的块分给那台机器的那个task去执行Map，即jobTracker如何调度任务给taskTracker？

本地化运行机制和jobTracker的调度算法；（可能是就近原则）.

数据本地化的(data-local),也就是任务运行在输入分片所在的节点上。同样，任务也可能是机器本地化的：任务和输入分片在同一个机架，但不在同一个节点上。reduce任务，jobtracker简单滴从待运行的reduce任务列表中选取下一个来运行，用不着考虑数据段饿本地化。

l  createRecordReader()按照必然格式读取响应数据：

              接收split块，返回读取记录的结果； 

public RecordReader<ImmutableBytesWritable,Result> createRecordReader(InputSplit split, TaskAttemptContext context){
 
                  
 
                }

                trr.init()返回的是这个分块的起始rowkey的记录;

               RecordReader将一个split解析成<key,value>对的形式提供给map函数，key就是rowkey，value就是对应的一行数据；

               RecordReader用于在划分中读取<Key,Value>对。RecordReader有五个虚方法，分别是： initialize：初始化，输入参数包括该Reader工作的数据划分InputSplit和Job的上下文context；nextKey：得到输 入的下一个Key，如果数据划分已经没有新的记录，返回空； nextValue：得到Key对应的Value，必须在调用nextKey后调用；getProgress：得到现在的进度； close：来自java.io的Closeable接口，用于清理RecordReader。

5.2   job.setInputFormatClass(TableInputFormat.class);         

5.3   TableMapReduceUtil.initTableReducerJob("daily_result",DailyReduce.class, job);

       使用了该方法就不需要再单独定义

       initTableReducerJob()方法完成了一系列操作：

                (1). job.setOutputFormatClass(TableOutputFormat.class); 设置输出格式；

                (2). conf.set(TableOutputFormat.OUTPUT_TABLE, table);设置输出表；

                     (3).初始化partition；

六：HBase测试点：

前提：自己维护HBase集群，否则无需关注HBase本身。

1.      功能测试：

(1)  Row Key的校验(重点)：

rowkey的长度、rowkey的排序、rowkey是否有遗失

(2)  Value的校验：

(3)  Table schema：

TTL(生存周期)：

压缩方式：Value值的压缩是否出错。

(4)  Family名称正确性的校验：

(5)  破坏性校验：

由于HBase的数据都是在集群中有备份的，所以才去人工宕机，查看数据是否能够正常取出。

2.     性能测试：

(1)  对HBase性能测试的工具：YCSB

YCSB（Yahoo!Cloud Serving Benchmark）是雅虎开源的一款通用的性能测试工具。

通过这个工具我们可以对各类NoSQL产品进行相关的性能测试。

参考文档：

七：Hadoop测试点：

1.    Job任务请求：

job需要解析一个request的请求文件，这里需要考虑到文件编码格式的问题。

2.    MR数据处理：

(1)  MR异常：

3.    程序的稳定和优化：

Hadoop测试参考：HADOOP测试常见问题和测试方法.docx

八：附

1.    RPC通信协议：

RPC（RemoteProcedure Call Protocol）——远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。RPC协议假定某些传输协议的 存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层。RPC使得开发包括网络分布式多程序在内的 应用程序更加容易。

九：随想：

1.    Hadoop 的分布式并行运算有一个作为主控的JobTracker，用于调度和管理其它的 TaskTracker, JobTracker 可以运行于集群中任一台计算机上。TaskTracker负责执行任务，必须运行于 DataNode 上，即 DataNode 既是数据存储结点，也是计算结点。

思考： JobTracker是如何从闲置的机器中选择出来的？是不是任何一台集群中的机器都可能有成为JobTracker的可能？所以机器都同事装了JobTracker和TaskTracker吗？

是谁在管理着JobTracker的分配和TaskTracker的运行？

2.     

随笔记录：

1.Zookeeper中记录了-ROOT-表的location，我们的程序会通过我们配置的zookeeper地址找到zookeeper，然 后根据zookeeper中存储的-ROOT-表的location，去到相应的机器上访问-ROOT-表，根据-ROOT-表中描述的.META表找到 相应的Ration信 息。                                                                   

-ROOT-表只有一个区域，而.META可以有多个区域。