Hadoop

 

HBase/Hadoop学习笔记【这是我转载的，原著地址：http://blog.csdn.net/yanmingming1989/article/details/8558880】

2013-01-31 11:25 20491人阅读 评论(4) 收藏 举报

 分类：

NoSql（2） 

 

 

 

 

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

文件具体配置为：

 

<configuration>
  <property>
     <name>hbase.zookeeper.property.clientPort</name>
      <value>2222</value>
   </property>
   <property>
     <name>hbase.zookeeper.quorum</name>
      <value>xxx.xxx.xxx.xxx.xxx</value>
   </property>
 <property skipInDoc="true">
 <name>hbase.defaults.for.version</name>
 <value>0.90.2</value>
  </property>
 </configuration>

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

 

4.    写测试文件：

public class HbaseTest {
       private static Configuration conf = null;
         static {
            conf = HBaseConfiguration.create();
           conf.addResource("conf/hbase-site.xml");
      }}

 

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

 

 

三、基础知识：

 

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工程加入依赖：

 

                  <dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase</artifactId>
<version>0.90.2</version>
</dependency>

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

 

       <repositories>
<repository>
<id>cloudera</id>
<url>https://repository.cloudera.com/content/groups/public</url>
</repository>
</repositories>

 

                     步骤二：

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

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

 

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

                                    http://www.cnblogs.com/panfeng412/archive/2011/08/14/2137984.html

 

 

四、深入理解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.    …..

参考：http://hi.baidu.com/surendaxiao/blog/item/cb1b42f86b03084e252df233.html

6.    HMaster：

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

　　1. 管理用户对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之间不能共享静态类属性，即使是在同一台机器上，因为是以进程的方式在运行。

 

 

        1. 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）。

                       Partitioner负责控制map输出结果key的分割。Key（或者一个key子集）被用 于产生分区，通常使用的是Hash函数。分区的数目与一个作业的reduce任务的数目是一样的。因此，它控制将中间过程的key（也就是这条记录）应该发送给m个reduce任务中的哪一个来进行reduce操作。

        

        2. 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的输出是一边被取回一边被合并的。

 

        3. 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“执行完成。

         安装还可以参考文档：http://blog.csdn.net/savechina/article/details/5656937

         按照“在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)写入文件系统

参考文档：http://blog.csdn.net/derekjiang/article/details/6851625

思考：

 

 

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作用？

            将scan实例转换为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？

                 A: 需要进一步了解Map的本地化运行机制和jobTracker的调度算法；（可能是就近原则）.

                    对于一个map任务，jobtracker会考虑tasktracker的网络位置，并选取一个距离其输入分片文件最近的tasktracker。在最理想的情况下，任务是数据本地化的(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产品进行相关的性能测试。

参考文档：http://www.cnblogs.com/gpcuster/archive/2011/08/16/2141430.html

 

参考：http://www.spnguru.com/(趋势科技)

 

 

七：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可以有多个区域。