HBase API应用和优化创建Maven工程，添加依赖<dependencies> <dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-client</artifactId> <version>1.3.1</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> <dependency> <groupId>org.testng</groupId> <artifactId>testng</artifactId> <version>6.14.3</version> <scope>test</scope> </dependency> </dependencies>测试类编写通用代码public class HbaseClientDemo { Configuration conf=null; Connection conn=null; HBaseAdmin admin =null; @Before public void init () throws IOException { conf = HBaseConfiguration.create(); conf.set("hbase.zookeeper.quorum","linux121,linux122"); conf.set("hbase.zookeeper.property.clientPort","2181"); conn = ConnectionFactory.createConnection(conf); } @After public void destroy(){ if(admin!=null){ try { admin.close(); } catch (IOException e) { e.printStackTrace(); } } if(conn !=null){ try { conn.close(); } catch (IOException e) { e.printStackTrace(); } } } }创建表：@Test public void createTable() throws IOException { admin = (HBaseAdmin) conn.getAdmin(); //创建表描述器 HTableDescriptor teacher = new HTableDescriptor(TableName.valueOf("teacher")); //设置列族描述器 teacher.addFamily(new HColumnDescriptor("info")); //执行创建操作 admin.createTable(teacher); System.out.println("teacher表创建成功！！"); }插入数据：//插入一条数据 @Test public void putData() throws IOException { //获取一个表对象 Table t = conn.getTable(TableName.valueOf("teacher")); //设定rowkey Put put = new Put(Bytes.toBytes("110")); //列族，列，value put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("addr"), Bytes.toBytes("beijing")); //执行插入 t.put(put); // t.put();//可以传入list批量插入数据 //关闭table对象 t.close(); System.out.println("插入成功！！"); }删除数据：//删除一条数据 @Test public void deleteData() throws IOException { //需要获取一个table对象 final Table worker = conn.getTable(TableName.valueOf("worker")); //准备delete对象 final Delete delete = new Delete(Bytes.toBytes("110")); //执行删除 worker.delete(delete); //关闭table对象 worker.close(); System.out.println("删除数据成功！！"); }查询某个列族数据：//查询某个列族数据 @Test public void getDataByCF() throws IOException { //获取表对象 HTable teacher = (HTable) conn.getTable(TableName.valueOf("teacher")); //创建查询的get对象 Get get = new Get(Bytes.toBytes("110")); //指定列族信息 // get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("sex")); get.addFamily(Bytes.toBytes("info")); //执行查询 Result res = teacher.get(get); Cell[] cells = res.rawCells(); //获取改行的所有cell对象 for (Cell cell : cells) { //通过cell获取rowkey,cf,column,value String cf = Bytes.toString(CellUtil.cloneFamily(cell)); String column = Bytes.toString(CellUtil.cloneQualifier(cell)); String value = Bytes.toString(CellUtil.cloneValue(cell)); String rowkey = Bytes.toString(CellUtil.cloneRow(cell)); System.out.println(rowkey + "----" + cf + "---" + column + "---" + value); } teacher.close(); //关闭表对象资源 }通过Scan全表扫描：//全表扫描 @Test public void scanAllData() throws IOException { HTable teacher = (HTable) conn.getTable(TableName.valueOf("teacher")); Scan scan = new Scan(); ResultScanner resultScanner = teacher.getScanner(scan); for (Result result : resultScanner) { Cell[] cells = result.rawCells(); //获取改行的所有cell对象 for (Cell cell : cells) { //通过cell获取rowkey,cf,column,value String cf = Bytes.toString(CellUtil.cloneFamily(cell)); String column = Bytes.toString(CellUtil.cloneQualifier(cell)); String value = Bytes.toString(CellUtil.cloneValue(cell)); String rowkey = Bytes.toString(CellUtil.cloneRow(cell)); System.out.println(rowkey + "----" + cf + "--" + column + "---" + value); } } teacher.close(); }通过startRowKey和endRowKey进行扫描：//通过startRowKey和endRowKey进行扫描查询 @Test public void scanRowKey() throws IOException { HTable teacher = (HTable) conn.getTable(TableName.valueOf("teacher")); Scan scan = new Scan(); scan.setStartRow("0001".getBytes()); scan.setStopRow("2".getBytes()); ResultScanner resultScanner = teacher.getScanner(scan); for (Result result : resultScanner) { Cell[] cells = result.rawCells(); //获取改行的所有cell对象 for (Cell cell : cells) { //通过cell获取rowkey,cf,column,value String cf = Bytes.toString(CellUtil.cloneFamily(cell)); String column = Bytes.toString(CellUtil.cloneQualifier(cell)); String value = Bytes.toString(CellUtil.cloneValue(cell)); String rowkey = Bytes.toString(CellUtil.cloneRow(cell)); System.out.println(rowkey + "----" + cf + "--" + column + "---" + value); } } teacher.close(); }Hbase 协处理器协处理器概述官方地址：http://hbase.apache.org/book.html#cp访问HBase的方式是使用scan或get获取数据，在获取到的数据上进行业务运算。但是在数据量非常大的时候，比如一个有上亿行及十万个列的数据集，再按常用的方式移动获取数据就会遇到性能问题。客户端也需要有强大的计算能力以及足够的内存来处理这么多的数据。此时就可以考虑使用Coprocessor(协处理器)。将业务运算代码封装到Coprocessor中并在RegionServer上运行，即在数据实际存储位置执行，最后将运算结果返回到客户端。利用协处理器，用户可以编写运行在 HBase Server 端的代码。Hbase Coprocessor类似以下概念触发器和存储过程：一个Observer Coprocessor有些类似于关系型数据库中的触发器，通过它我们可以在一些事件（如Get或是Scan）发生前后执行特定的代码。Endpoint Coprocessor则类似于关系型数据库中的存储过程，因为它允许我们在RegionServer上直接对它存储的数据进行运算，而非是在客户端完成运算。MapReduce：MapReduce的原则就是将运算移动到数据所处的节点。Coprocessor也是按照相同的原则去工作的。AOP：如果熟悉AOP的概念的话，可以将Coprocessor的执行过程视为在传递请求的过程中对请求进行了拦截，并执行了一些自定义代码。协处理器类型Observer协处理器与触发器(trigger)类似：在一些特定事件发生时回调函数（也被称作钩子函数，hook）被执行。这些事件包括一些用户产生的事件，也包括服务器端内部自动产生的事件。协处理器框架提供的接口如下：RegionObserver：用户可以用这种的处理器处理数据修改事件，它们与表的region联系紧密。MasterObserver：可以被用作管理或DDL类型的操作，这些是集群级事件。WALObserver：提供控制WAL的钩子函数Endpoint这类协处理器类似传统数据库中的存储过程，客户端可以调用这些 Endpoint 协处理器在Regionserver中执行一段代码，并将 RegionServer 端执行结果返回给客户端进一步处理。Endpoint常见用途聚合操作假设需要找出一张表中的最大数据，即 max 聚合操作，普通做法就是必须进行全表扫描，然后Client代码内遍历扫描结果，并执行求最大值的操作。这种方式存在的弊端是无法利用底层集群的并发运算能力，把所有计算都集中到 Client 端执行，效率低下。使用Endpoint Coprocessor，用户可以将求最大值的代码部署到 HBase RegionServer 端，HBase 会利用集群中多个节点的优势来并发执行求最大值的操作。也就是在每个 Region 范围内执行求最大值的代码，将每个 Region 的最大值在 Region Server 端计算出，仅仅将该 max 值返回给Client。在Client进一步将多个 Region 的最大值汇总进一步找到全局的最大值。Endpoint Coprocessor的应用后续可以借助于Phoenix非常容易就能实现。针对Hbase数据集进行聚合运算直接使用SQL语句就能搞定。Observer 案例需求：通过协处理器Observer实现Hbase当中t1表插入数据，指定的另一张表t2也需要插入相对应的数据。create 't1','info' create 't2','info'实现思路：通过Observer协处理器捕捉到t1插入数据时，将数据复制一份并保存到t2表中开发步骤：添加依赖<dependency> <groupId>org.apache.hbase</groupId> <artifactId>hbase-server</artifactId> <version>1.3.1</version> </dependency>编写Observer协处理器package com.test.coprocessor; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CoprocessorEnvironment; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver; import org.apache.hadoop.hbase.coprocessor.ObserverContext; import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; import org.apache.hadoop.hbase.filter.ByteArrayComparable; import org.apache.hadoop.hbase.filter.CompareFilter; import org.apache.hadoop.hbase.regionserver.wal.WALEdit; import java.io.IOException; public class MyProcessor extends BaseRegionObserver { @Override public void prePut(ObserverContext<RegionCoprocessorEnvironment> ce, Put put, WALEdit edit, Durability durability) throws IOException { //把自己需要执行的逻辑定义在此处，向t2表插入数据，数据具体是什么内容与Put一样 final HTableInterface t2 = e.getEnvironment().getTable(TableName.valueOf("t2")); //解析t1表的插入对象put final Cell cell = put.get(Bytes.toBytes("info"), Bytes.toBytes("name")).get(0); //table对象.put final Put put1 = new Put(put.getRow()); put1.add(cell); t2.put(put1); //执行向t2表插入数据 t2.close(); } }打成Jar包，上传HDFScd /opt/test/softwares mv original-hbaseStudy-1.0-SNAPSHOT.jar processor.jar hdfs dfs -mkdir -p /processor hdfs dfs -put processor.jar /processor挂载协处理器describe 't1' alter 't1',METHOD => 'table_att','Coprocessor'=>'hdfs://linux121:9000/processor/processor.jar|com.test.hbase.processor.MyProcessor|1001|' describe 't1'1001表示优先级，在多个协处理器存在的情况下，优先级越小越优先验证协处理器向t1表中插入数据(shell方式验证)put 't1','rk1','info:name','lisi'卸载协处理器disable 't1' alter 't1',METHOD=>'table_att_unset',NAME=>'coprocessor$1' enable 't2'HBase表的RowKey设计RowKey的基本介绍ASCII码字典顺序。012,0,123,234,3.0,3,012,123,2340,012,123,234,3字典序的排序规则。先比较第一个字节，如果相同，然后比对第二个字节，以此类推，如果到第X个字节，其中一个已经超出了rowkey的长度，短rowkey排在前面。RowKey长度原则rowkey是一个二进制码流，可以是任意字符串，最大长度64kb，实际应用中一般为10-100bytes，以byte[]形式保存，一般设计成定长。建议越短越好，不要超过16个字节，设计过长会降低memstore内存的利用率和HFile存储数据的效率。RowKey散列原则建议将rowkey的高位作为散列字段，这样将提高数据均衡分布在每个RegionServer，以实现负载均衡的几率。RowKey唯一原则必须在设计上保证其唯一性，访问hbase table中的行：有3种方式：单个rowkeyrowkey 的range全表扫描(一定要避免全表扫描)实现方式：1）org.apache.hadoop.hbase.client.Get2）scan方法： org.apache.hadoop.hbase.client.Scanscan使用的时候注意：setStartRow，setEndRow 限定范围，范围越小，性能越高。RowKey排序原则HBase的Rowkey是按照ASCII有序设计的，我们在设计Rowkey时要充分利用这点。HBase表的热点什么是热点检索habse的记录首先要通过row key来定位数据行。当大量的client访问hbase集群的一个或少数几个节点，造成少数region server的读/写请求过多、负载过大，而其他region server负载却很小，就造成了“热点”现象。热点的解决方案预分区预分区的目的让表的数据可以均衡的分散在集群中，而不是默认只有一个region分布在集群的一个节点上。加盐这里所说的加盐不是密码学中的加盐，而是在rowkey的前面增加随机数，具体就是给rowkey分配一个随机前缀以使得它和之前的rowkey的开头不同。4个region，[,a),[a,b),[b,c),[c,]原始数据：abc1,abc2,abc3.加盐后的rowkey：a-abc1,b-abc2,c-abc3abc1,aabc2,b哈希哈希会使同一行永远用一个前缀加盐。哈希也可以使负载分散到整个集群，但是读却是可以预测的。使用确定的哈希可以让客户端重构完整的rowkey，可以使用get操作准确获取某一个行数据。原始数据： abc1，abc2,abc3哈希：md5（abc1）=92231b....., 9223-abc1md5（abc2）=32a131122...., 32a1-abc2md5（abc3）= 452b1...., 452b-abc3反转反转固定长度或者数字格式的rowkey。这样可以使得rowkey中经常改变的部分（最没有意义的部分）放在前面。这样可以有效的随机rowkey，但是牺牲了rowkey的有序性。15X，13X反转后为X51，X31HBase的二级索引HBase表按照rowkey查询性能是最高的。rowkey就相当于hbase表的一级索引。为了HBase的数据查询更高效、适应更多的场景，诸如使用非rowkey字段检索也能做到秒级响应，或者支持各个字段进行模糊查询和多字段组合查询等， 因此需要在HBase上面构建二级索引，以满足现实中更复杂多样的业务需求。hbase的二级索引其本质就是建立hbase表中列与行键之间的映射关系。常见的二级索引我们一般可以借助各种其他的方式来实现，例如Phoenix或者solr或者ES等。布隆过滤器在HBase的应用布隆过滤器应用之前在讲HBase的数据存储原理的时候，我们知道HBase的读操作需要访问大量的文件，大部分的实现通过布隆过滤器来避免大量的读文件操作。布隆过滤器的原理通常判断某个元素是否存在用的可以选择hashmap。但是 HashMap 的实现也有缺点，例如存储容量占比高，考虑到负载因子的存在，通常空间是不能被用满的，而一旦你的值很多例如上亿的时候，那 HashMap 占据的内存大小就变得很可观了。Bloom Filter是一种空间效率很高的随机数据结构，它利用位数组很简洁地表示一个集合，并能判断一个元素是否属于这个集合。hbase中布隆过滤器来过滤指定的rowkey是否在目标文件，避免扫描多个文件。使用布隆过滤器来判断。布隆过滤器返回true，结果不一定正确，如果返回false，则说明确实不存在。原理示意图Bloom Filter案例布隆过滤器，已经不需要自己实现，Google已经提供了非常成熟的实现。导入依赖<dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>27.0.1-jre</version> </dependency>使用guava 的布隆过滤器，封装的非常好，使用起来非常简洁方便。例：预估数据量1w，错误率需要减小到万分之一。使用如下代码进行创建。public static void main(String[] args) { // 1.创建符合条件的布隆过滤器 // 预期数据量10000，错误率0.0001 BloomFilter<CharSequence> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")),10000, 0.0001); // 2.将一部分数据添加进去 for (int i = 0; i < 5000; i++) { bloomFilter.put("" + i); } System.out.println("数据写入完毕"); // 3.测试结果 for (int i = 0; i < 10000; i++) { if (bloomFilter.mightContain("" + i)) { System.out.println(i + "存在"); } else { System.out.println(i + "不存在"); } } }
HBase读数据流程HBase读操作1）首先从zk找到meta表的region位置，然后读取meta表中的数据，meta表中存储了用户表的region信息2）根据要查询的namespace、表名和rowkey信息。找到写入数据对应的region信息3）找到这个region对应的regionServer，然后发送请求4）查找对应的region5）先从memstore查找数据，如果没有，再从BlockCache上读取HBase上Regionserver的内存分为两个部分一部分作为Memstore，主要用来写；另外一部分作为BlockCache，主要用于读数据；6）如果BlockCache中也没有找到，再到StoreFile上进行读取从storeFile中读取到数据之后，不是直接把结果数据返回给客户端，而是把数据先写入到BlockCache中，目的是为了加快后续的查询；然后在返回结果给客户端。HBase写数据流程HBase写操作1）首先从zk找到meta表的region位置，然后读取meta表中的数据，meta表中存储了用户表的region信息2）根据namespace、表名和rowkey信息。找到写入数据对应的region信息3）找到这个region对应的regionServer，然后发送请求4）把数据分别写到HLog（write ahead log）和memstore各一份5）memstore达到阈值后把数据刷到磁盘，生成storeFile文件6）删除HLog中的历史数据HBase的flush(刷写)及compact(合并)机制Flush机制（1）当memstore的大小超过这个值的时候，会flush到磁盘，默认为128M<property> <name>hbase.hregion.memstore.flush.size</name> <value>134217728</value> </property>（2）当memstore中的数据时间超过1小时，会flush到磁盘<property> <name>hbase.regionserver.optionalcacheflushinterval</name> <value>3600000</value> </property>（3）HregionServer的全局memstore的大小，超过该大小会触发flush到磁盘的操作,默认是堆大小的40%<property> <name>hbase.regionserver.global.memstore.size</name> <value>0.4</value> </property>（4）手动flushflush tableName阻塞机制以上介绍的是Store中memstore数据刷写磁盘的标准，但是Hbase中是周期性的检查是否满足以上标准满足则进行刷写，但是如果在下次检查到来之前，数据疯狂写入Memstore中，会出现什么问题？会触发阻塞机制，此时无法写入数据到Memstore，数据无法写入Hbase集群。memstore中数据达到512MB计算公式：hbase.hregion.memstore.flush.size * hbase.hregion.memstore..block.multiplierhbase.hregion.memstore.flush.size刷写的阀值，默认是 134217728，即128MB。hbase.hregion.memstore.block.multiplier是一个倍数，默认是4。RegionServer全部memstore达到规定值hbase.regionserver.global.memstore.size.lower.limit是0.95，hbase.regionserver.global.memstore.size是0.4，堆内存总共是 16G，触发刷写的阈值是：6.08GB触发阻塞的阈值是：6.4GBCompact合并机制在hbase中主要存在两种类型的compac合并minor compact小合并在将Store中多个HFile(StoreFile)合并为一个HFile这个过程中，删除和更新的数据仅仅只是做了标记，并没有物理移除，这种合并的触发频率很高。minor compact文件选择标准由以下几个参数共同决定： <property> <name>hbase.hstore.compaction.min</name> <value>3</value> </property> <property> <name>hbase.hstore.compaction.max</name> <value>10</value> </property> <property> <name>hbase.hstore.compaction.min.size</name> <value>134217728</value> </property> <property> <name>hbase.hstore.compaction.max.size</name> <value>9223372036854775807</value> </property>触发条件memstore flush在进行memstore flush前后都会进行判断是否触发compact定期检查线程周期性检查是否需要进行compaction操作,由参数：hbase.server.thread.wakefrequency决定，默认值是10000 millsecondsmajor compact大合并合并Store中所有的HFile为一个HFile这个过程有删除标记的数据会被真正移除，同时超过单元格maxVersion的版本记录也会被删除。合并频率比较低，默认7天执行一次，并且性能消耗非常大，建议生产关闭（设置为0），在应用空闲时间手动触发。一般可以是手动控制进行合并，防止出现在业务高峰期。major compaction触发时间条件 <property> <name>hbase.hregion.majorcompaction</name> <value>604800000</value> </property>手动触发，使用major_compact命令major_compact tableNameRegion 拆分机制Region中存储的是大量的rowkey数据，当Region中的数据条数过多的时候，直接影响查询效率。当Region过大的时候.HBase会拆分Region，这也是Hbase的一个优点 。拆分策略HBase的Region Split策略一共有以下几种：1）ConstantSizeRegionSplitPolicy0.94版本前默认切分策略当region大小大于某个阈值(hbase.hregion.max.filesize=10G)之后就会触发切分，一个region等分为2个region。但是在生产线上这种切分策略却有相当大的弊端：切分策略对于大表和小表没有明显的区分。阈值(hbase.hregion.max.filesize)设置较大对大表比较友好，但是小表就有可能不会触发分裂，极端情况下可能就1个，这对业务来说并不是什么好事。如果设置较小则对小表友好，但一个大表就会在整个集群产生大量的region，这对于集群的管理、资源使用、failover来说都不是一件好事。2）IncreasingToUpperBoundRegionSplitPolicy0.94版本~2.0版本默认切分策略切分策略稍微有点复杂，总体看和ConstantSizeRegionSplitPolicy思路相同，一个region大小大于设置阈值就会触发切分。但是这个阈值并不像ConstantSizeRegionSplitPolicy是一个固定的值，而是会在一定条件下不断调整，调整规则和region所属表在当前regionserver上的region个数有关系。region split的计算公式是：regioncount^3 128M 2，当region达到该size的时候进行split例如：第一次split：1^3 * 256 = 256MB第二次split：2^3 * 256 = 2048MB第三次split：3^3 * 256 = 6912MB第四次split：4^3 * 256 = 16384MB > 10GB，因此取较小的值10GB后面每次split的size都是10GB了3）SteppingSplitPolicy2.0版本默认切分策略这种切分策略的切分阈值又发生了变化，相比 IncreasingToUpperBoundRegionSplitPolicy 简单了一些，依然和待分裂region所属表在当前regionserver上的region个数有关系，如果region个数等于1，切分阈值为flush size * 2，否则为MaxRegionFileSize。这种切分策略对于大集群中的大表、小表会比IncreasingToUpperBoundRegionSplitPolicy 更加友好，小表不会再产生大量的小region，而是适可而止。4）KeyPrefixRegionSplitPolicy根据rowKey的前缀对数据进行分组，这里是指定rowKey的前多少位作为前缀，比如rowKey都是16位的，指定前5位是前缀，那么前5位相同的rowKey在进行region split的时候会分到相同的region中。5）DelimitedKeyPrefixRegionSplitPolicy保证相同前缀的数据在同一个region中，例如rowKey的格式为：userid_eventtype_eventid，指定的delimiter为 _ ，则split的的时候会确保userid相同的数据在同一个region中。6）DisabledRegionSplitPolicy不启用自动拆分，需要指定手动拆分。RegionSplitPolicy的应用Region拆分策略可以全局统一配置，也可以为单独的表指定拆分策略。1）通过hbase-site.xml全局统一配置（对hbase所有表生效）<property> <name>hbase.regionserver.region.split.policy</name> <value>org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy</value> </property>2）通过Java API为单独的表指定Region拆分策略HTableDescriptor tableDesc = new HTableDescriptor("test1"); tableDesc.setValue(HTableDescriptor.SPLIT_POLICY, IncreasingToUpperBoundRegionSplitPolicy.class.getName()); tableDesc.addFamily(new HColumnDescriptor(Bytes.toBytes("cf1"))); admin.createTable(tableDesc);3）通过HBase Shell为单个表指定Region拆分策略create 'test2', {METADATA => {'SPLIT_POLICY' => 'org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy'}},{NAME => 'cf1'}HBase表的预分区（region）为何要预分区？当一个table刚被创建的时候，Hbase默认的分配一个region给table。也就是说这个时候，所有的读写请求都会访问到同一个regionServer的同一个region中，这个时候就达不到负载均衡的效果了，集群中的其他regionServer就可能会处于比较空闲的状态。解决这个问题可以用pre-splitting,在创建table的时候就配置好，生成多个region。增加数据读写效率负载均衡，防止数据倾斜方便集群容灾调度region每一个region维护着startRow与endRowKey，如果加入的数据符合某个region维护的rowKey范围，则该数据交给这个region维护手动指定预分区create 'person','info1','info2',SPLITS => ['1000','2000','3000']也可以把分区规则创建于文件中vim split.txt文件内容aaa bbb ccc ddd执行create 'student','info',SPLITS_FILE => '/root/hbase/split.txt'Region 合并Region合并说明Region的合并不是为了性能，而是出于维护的目的。如何进行Region合并通过Merge类冷合并Region需要先关闭hbase集群需求：需要把student表中的2个region数据进行合并：student,,1593244870695.10c2df60e567e73523a633f20866b4b5.student,1000,1593244870695.0a4c3ff30a98f79ff6c1e4cc927b3d0d.这里通过org.apache.hadoop.hbase.util.Merge类来实现，不需要进入hbase shell，直接执行（需要先关闭hbase集群）：hbase org.apache.hadoop.hbase.util.Merge student student,,1595256696737.fc3eff4765709e66a8524d3c3ab42d59. student,aaa,1595256696737.1d53d6c1ce0c1bed269b16b6514131d0.通过online_merge热合并Region不需要关闭hbase集群，在线进行合并与冷合并不同的是，online_merge的传参是Region的hash值，而Region的hash值就是Region名称的最后那段在两个.之间的字符串部分。需求：需要把student表中的2个region数据进行合并：student,,1587392159085.9ca8689901008946793b8d5fa5898e06.student,aaa,1587392159085.601d5741608cedb677634f8f7257e000.需要进入hbase shell：merge_region 'c8bc666507d9e45523aebaffa88ffdd6','02a9dfdf6ff42ae9f0524a3d8f4c7777'成功后观察界面
HBase 简介HBase是什么HBase 基于 Google的BigTable论文而来，是一个分布式海量列式非关系型数据库系统，可以提供超大规模数据集的实时随机读写。接下来，通过一个场景初步认识HBase列存储如下MySQL存储机制，空值字段浪费存储空间idNAMEAGESALARYJOB1小明23 学生2小红 10w律师 如果是列存储的话，可以这么玩......rowkey：1 name：小明rowkey：1 age：23rowkey：1 job：学生rowkey：2 name ：小红rowkey：2 salary：10wrowkey：2 job：律师....列存储的优点：1）减少存储空间占用。2）支持好多列HBase的特点海量存储：底层基于HDFS存储海量数据列式存储：HBase表的数据是基于列族进行存储的，一个列族包含若干列极易扩展：底层依赖HDFS，当磁盘空间不足的时候，只需要动态增加DataNode服务节点就可以高并发：支持高并发的读写请求稀疏：稀疏主要是针对HBase列的灵活性，在列族中，你可以指定任意多的列，在列数据为空的情况下，是不会占用存储空间的。数据的多版本：HBase表中的数据可以有多个版本值，默认情况下是根据版本号去区分，版本号就是插入数据的时间戳数据类型单一：所有的数据在HBase中是以字节数组进行存储HBase的应用交通方面：船舶GPS信息，每天有上千万左右的数据存储。金融方面：消费信息、贷款信息、信用卡还款信息等电商方面：电商网站的交易信息、物流信息、游览信息等电信方面：通话信息总结：HBase适合海量明细数据的存储，并且后期需要有很好的查询性能（单表超千万、上亿，且并发要求高）HBase数据模型HBase的数据也是以表（有行有列）的形式存储HBase逻辑架构HBase物理存储概念描述NameSpace（数据库）命名空间，类似于关系型数据库的database概念，每个命名空间下有多个表。HBase两个自带的命名空间，分别是hbase和default，hbase中存放的是HBase内置的表，default表是用户默认使用的命名空间。一个表可以自由选择是否有命名空间，如果创建表的时候加上了命名空间后，这个表名字以:作为区分。Table类似于关系型数据库的表概念。不同的是，HBase定义表时只需要声明列族即可，数据属性，比如超时时间（TTL），压缩算法（COMPRESSION）等，都在列族的定义中定义，不需要声明具体的列。Row（一行逻辑数据）HBase表中的每行数据都由一个RowKey和多个Column（列）组成。一个行包含了多个列，这些列通过列族来分类,行中的数据所属列族只能从该表所定义的列族中选取，不能定义这个表中不存在的列族，否则报错NoSuchColumnFamilyException。RowKey（每行数据主键）Rowkey由用户指定的一串不重复的字符串定义，是一行的唯一标识！数据是按照RowKey的字典顺序存储的，并且查询数据时只能根据RowKey进行检索，所以RowKey的设计十分重要。如果使用了之前已经定义的RowKey，那么会将之前的数据更新掉。Column Family（列族）列族是多个列的集合。一个列族可以动态地灵活定义多个列。表的相关属性大部分都定义在列族上，同一个表里的不同列族可以有完全不同的属性配置，但是同一个列族内的所有列都会有相同的属性。列族存在的意义是HBase会把相同列族的列尽量放在同一台机器上，所以说，如果想让某几个列被放到一起，你就给他们定义相同的列族。Column Qualifier（列）Hbase中的列是可以随意定义的，一个行中的列不限名字、不限数量，只限定列族。因此列必须依赖于列族存在！列的名称前必须带着其所属的列族！例如info：name，info：ageTimeStamp（时间戳→版本）用于标识数据的不同版本（version）。时间戳默认由系统指定，也可以由用户显式指定。在读取单元格的数据时，版本号可以省略，如果不指定，Hbase默认会获取最后一个版本的数据返回。Cell一个列中可以存储多个版本的数据。而每个版本就称为一个单元格（Cell）。Region（表的分区）Region由一个表的若干行组成！在Region中行的排序按照行键（rowkey）字典排序。Region不能跨RegionSever，且当数据量大的时候，HBase会拆分Region。HBase整体架构Zookeeper实现了HMaster的高可用保存了HBase的元数据信息，是所有HBase表的寻址入口对HMaster和HRegionServer实现了监控HMaster（Master）为HRegionServer分配Region维护整个集群的负载均衡维护集群的元数据信息发现失效的Region，并将失效的Region分配到正常的HRegionServer上HRegionServer（RegionServer）负责管理Region接受客户端的读写数据请求切分在运行过程中变大的RegionRegion每个HRegion由多个Store构成，每个Store保存一个列族（Columns Family），表有几个列族，则有几个Store，每个Store由一个MemStore和多个StoreFile组成，MemStore是Store在内存中的内容，写到文件后就是StoreFile。StoreFile底层是以HFile的格式保存。HBase集群安装部署（1）下载安装包http://archive.apache.org/dist/hbase/1.3.1/hbase-1.3.1-bin.tar.gz（2）规划安装目录/opt/test/servers/（3）上传安装包到服务器（4）解压安装包到指定的规划目录tar -zxvf hbase-1.3.1-bin.tar.gz -C /opt/test/servers（5）修改配置文件需要把hadoop中的配置core-site.xml 、hdfs-site.xml拷贝到hbase安装目录下的conf文件夹中（建议采用软链接）ln -s /opt/test/servers/hadoop-2.9.2/etc/hadoop/core-site.xml /opt/test/servers/hbase-1.3.1/conf/core-site.xml ln -s /opt/test/servers/hadoop-2.9.2/etc/hadoop/hdfs-site.xml /opt/test/servers/hbase-1.3.1/conf/hdfs-site.xml修改conf目录下配置文件修改 hbase-env.sh#添加java环境变量 export JAVA_HOME=/opt/module/jdk1.8.0_231 #指定使用外部的zk集群 export HBASE_MANAGES_ZK=FALSE修改 hbase-site.xml<configuration> <property> <name>hbase.rootdir</name> <value>hdfs://linux121:9000/hbase</value> </property> <property> <name>hbase.cluster.distributed</name> <value>true</value> </property> <property> <name>hbase.zookeeper.quorum</name> <value>linux121:2181,linux122:2181,linux123:2181</value> </property> </configuration>修改regionservers文件，指定regionserver节点linux121 linux122 linux123hbase的conf目录下创建文件backup-masters (Standby Master)linux122（6）配置hbase的环境变量（修改/etc/profile）export HBASE_HOME=/opt/test/servers/hbase-1.3.1 export PATH=$PATH:$HBASE_HOME/bin（7）分发hbase目录和环境变量到其他节点rsync-script hbase-1.3.1 rsync-script /etc/profile（8）让所有节点的hbase环境变量生效在所有节点执行 source /etc/profileHBase集群的启动和停止前提条件：先启动hadoop和zk集群启动HBase：start-hbase.sh停止HBase：stop-hbase.shHBase集群的web管理界面启动好HBase集群之后，可以访问地址：HMaster的主机名:16010HBase shell基本操作进入Hbase客户端命令操作界面hbase shell查看帮助命令help查看当前数据库中有哪些表list创建一张test表， 包含base_info、extra_info两个列族(Hbase建表必须指定列族信息)create 'test', 'base_info', 'extra_info'或者create 'test', {NAME => 'base_info', VERSIONS => '3'},{NAME => 'extra_info',VERSIONS => '3'}VERSIONS是指此单元格内的数据可以保留最近的3个版本添加数据操作向test表中插入信息，row key为rk1，列族base_info中添加name列标示符，值为wangput 'test', 'rk1', 'base_info:name', 'wang'向test表中插入信息，row key为rk1，列族base_info中添加age列标示符，值为30put 'test', 'rk1', 'base_info:age', 30向test表中插入信息，row key为rk1，列族extra_info中添加address列标示符，值为shanghaiput 'test', 'rk1', 'extra_info:address', 'shanghai'查询数据6.1 通过rowkey进行查询get 'test', 'rk1'6.2 查看rowkey下面的某个列族的信息获取test表中row key为rk1，base_info列族的所有信息get 'test', 'rk1', 'base_info'6.3 查看rowkey指定列族指定字段的值获取表中row key为rk1，base_info列族的name、age列标示符的信息get 'test', 'rk1', 'base_info:name', 'base_info:age'6.4 查看rowkey指定多个列族的信息获取test表中row key为rk1，base_info、extra_info列族的信息get 'test', 'rk1', 'base_info', 'extra_info'或者get 'test', 'rk1', {COLUMN => ['base_info', 'extra_info']}或者get 'test', 'rk1', {COLUMN => ['base_info:name', 'extra_info:address']}6.5 指定rowkey与列值查询获取表中row key为rk1，cell的值为wang的信息get 'test', 'rk1', {FILTER => "ValueFilter(=, 'binary:wang')"}6.6 指定rowkey与列值模糊查询获取表中row key为rk1，列标示符中含有a的信息get 'test', 'rk1', {FILTER => "(QualifierFilter(=,'substring:a'))"}6.7 查询所有数据查询test表中的所有信息scan 'test'6.8 列族查询查询表中列族为 base_info 的信息scan 'test', {COLUMNS => 'base_info'}或者scan 'test', {COLUMNS => 'base_info', RAW => true, VERSIONS => 3}注：Scan时可以设置是否开启Raw模式，开启Raw模式会返回包括已添加删除标记但是未实际删除的数据注：VERSIONS指定查询的最大版本数6.9 指定多个列族与按照数据值模糊查询查询test表中列族为 base_info 和 extra_info且列标示符中含有a字符的信息scan 'test', {COLUMNS => ['base_info', 'extra_info'], FILTER => "(QualifierFilter(=,'substring:a'))"}6.10 rowkey的范围值查询（非常重要）查询test表中列族为base_info，rk范围是[rk1, rk3)的数据（rowkey底层存储是字典序）按rowkey顺序存储。scan 'test', {COLUMNS => 'base_info', STARTROW => 'rk1', ENDROW => 'rk3'}6.11 指定rowkey模糊查询查询test表中row key以rk字符开头的scan 'test',{FILTER=>"PrefixFilter('rk')"}更新数据更新操作同插入操作一模一样，只不过有数据就更新，没数据就添加把test表中rowkey为rk1的base_info列族下的列name修改为liangput 'test', 'rk1', 'base_info:name', 'liang'删除数据和表8.1 指定rowkey以及列名进行删除删除test表row key为rk1，列标示符为 base_info:name 的数据delete 'test', 'rk1', 'base_info:name'8.2 指定rowkey，列名以及时间戳信息进行删除删除test表row key为rk1，列标示符为base_info:name的数据delete 'test', 'rk1', 'base_info:name', 16006606196558.3 删除列族删除 base_info 列族alter 'test', 'delete' => 'base_info'8.4 清空表数据删除test表数据truncate 'test'8.5 删除表删除test表（先disable 再drop）disable 'test' drop 'test'如果不进行disable，直接drop会报错ERROR: Table user is enabled. Disable it first.
Hadoop3.x中增强了很多特性，在Hadoop3.x中，不再允许使用jdk1.7，要求jdk1.8以上版本。这是因为Hadoop 2.0是基于JDK 1.7开发的，而JDK 1.7在2015年4月已停止更新，这直接迫使Hadoop社区基于JDK 1.8重新发布一个新的Hadoop版本，而这正是Hadoop3.x。Hadoop3.x以后将会调整方案架构，将Mapreduce 基于内存+io+磁盘，共同处理数据。Hadoop 3.x中引入了一些重要的功能和优化，包括HDFS 可擦除编码、多Namenode支持、MR Native Task优化、YARN基于cgroup的内存和磁盘IO隔离、YARN container resizing等。Hadoop3.x官方文档地址如下：http://hadoop.apache.org/docs/r3.0.1/Hadoop3.x新特性之Common改进Hadoop Common改进：精简Hadoop内核，包括剔除过期的API和实现，将默认组件实现替换成最高效的实现（比如将FileOutputCommitter缺省实现换为v2版本，废除hftp转由webhdfs替代，移除Hadoop子实现序列化库org.apache.hadoop.Recordslasspath isolation以防止不同版本jar包冲突，比如google Guava在混合使用Hadoop、HBase和Spark时，很容易产生冲突。（https://issues.apache.org/jira/browse/HADOOP-11656）Shell脚本重构。 Hadoop 3.0对Hadoop的管理脚本进行了重构，修复了大量bug，增加了新特性，支持动态命令等。使用方式上则和之前版本的一致。（https://issues.apache.org/jira/browse/HADOOP-9902）Hadoop3.x新特性之HDFS改进Hadoop3.x中最大改变的是HDFS，HDFS通过最近black块计算，根据最近计算原则，本地black块，加入到内存，先计算，通过IO，共享内存计算区域，最后快速形成计算结果。HDFS支持数据的擦除编码，这使得HDFS在不降低可靠性的前提下，节省一半存储空间。（https://issues.apache.org/jira/browse/HDFS-7285）多NameNode支持，即支持一个集群中，一个active、多个standby namenode部署方式。注：多ResourceManager特性在hadoop 2.0中已经支持。（https://issues.apache.org/jira/browse/HDFS-6440）关于这两个特性的官方文档地址：http://hadoop.apache.org/docs/r3.0.1/hadoop-project-dist/hadoophdfs/HDFSErasureCoding.html http://hadoop.apache.org/docs/r3.0.1/hadoop-project-dist/hadoophdfs/HDFSHighAvailabilityWithQJM.htmlHadoop3.x新特性之YARN改进基于cgroup的内存隔离和IO Disk隔离（https://issues.apache.org/jira/browse/YARN-2619）用curator实现RM leader选举（https://issues.apache.org/jira/browse/YARN-4438）containerresizing（https://issues.apache.org/jira/browse/YARN-1197）Timelineserver next generation （https://issues.apache.org/jira/browse/YARN-2928）官方文档地址：http://hadoop.apache.org/docs/r3.0.1/hadoop-yarn/hadoop-yarnsite/TimelineServiceV2.htmlHadoop3.x新特性之MapReduce改进Tasknative优化。为MapReduce增加了C/C++的map output collector实现（包括Spill，Sort和IFile等），通过作业级别参数调整就可切换到该实现上。对于shuffle密集型应用，其性能可提高约30%。（https://issues.apache.org/jira/browse/MAPREDUCE-2841）MapReduce内存参数自动推断。在Hadoop 2.0中，为MapReduce作业设置内存参数非常繁琐，涉及到两个参数：mapreduce.{map,reduce}.memory.mb和mapreduce.{map,reduce}.java.opts，一旦设置不合理，则会使得内存资源浪费严重，比如将前者设置为4096MB，但后者却是“-Xmx2g”，则剩余2g实际上无法让java heap使用到。（https://issues.apache.org/jira/browse/MAPREDUCE-5785）Hadoop3.x新特性之其他添加新的 hadoop-client-api 和 hadoop-client-runtime 组件到一个单独的jar包里，以此解决依赖不兼容的问题。（https://issues.apache.org/jira/browse/HADOOP-11804）支持微软的Azure分布式文件系统和阿里的aliyun分布式文件系统
源码阅读准备下载Apache Hadoop-2.9.2官方源码将源码导入idea中启动idea在提示界面选择导入点击Finish等待下载和解决依赖完成，源码导入成功！！NameNode 启动流程命令启动Hdfs集群start-dfs.sh该命令会启动Hdfs的NameNode以及DataNode，启动NameNode主要是通过org.apache.hadoop.hdfs.server.namenode.NameNode类。重点关注NameNode在启动过程中做了哪些工作（偏离主线的技术细节不深究）对于分析启动流程主要关注两部分代码：public class NameNode extends ReconfigurableBase implements NameNodeStatusMXBean { //该静态代码块主要是初始化一些HDFS的配置信息 static{ HdfsConfiguration.init(); //进入之后发现方法是空的，没有任何操作？其实不是观察HdfsConfiguration的静态代码块 } //HdfsConfiguration的类以及静态代码块 public class HdfsConfiguration extends Configuration { static { addDeprecatedKeys(); // adds the default resources Configuration.addDefaultResource("hdfs-default.xml"); Configuration.addDefaultResource("hdfs-site.xml"); } ... //main方法 public static void main(String argv[]) throws Exception{ //分析传入的参数是否为帮助参数，如果是帮助的话打印帮助信息，并退出。 if(DFSUtil.parseHelpArgument(argv,NameNode.USAGE,System.out,true)){ System.exit(0); } try{ //格式化输出启动信息，并且创建hook（打印节点关闭信息） StringUtils.startupShutdownMessage(NameNode.class,argv,LOG); //创建namenode NameNode namenode=createNameNode(argv,null); if（namenode!=null){ //加入集群 namenode.join(); } } catch(Throwable e){ //异常处理 LOG.error("Failed to start namenode.",e); terminate(1,e); } } ---------------------------------------------------------------------- //关注createNameNode public static NameNode createNameNode(String argv[], Configuration conf) throws IOException { LOG.info("createNameNode " + Arrays.asList(argv)); if (conf == null) conf = new HdfsConfiguration(); // Parse out some generic args into Configuration. GenericOptionsParser hParser = new GenericOptionsParser(conf, argv); argv = hParser.getRemainingArgs(); // Parse the rest, NN specific args. //解析启动的参数 StartupOption startOpt = parseArguments(argv); if (startOpt == null) { printUsage(System.err); return null; } setStartupOption(conf, startOpt); switch (startOpt) { .... default: { //正常启动进入该分支 //初始化metric系统 DefaultMetricsSystem.initialize("NameNode"); //返回新的NameNode return new NameNode(conf); } } } ---------------------------------------------------------------------- //NameNode的构造 public NameNode(Configuration conf) throws IOException { this(conf, NamenodeRole.NAMENODE); } ... protected NameNode(Configuration conf, NamenodeRole role) throws IOException { this.conf = conf; this.role = role; // 设置NameNode#clientNamenodeAddress为"hdfs://localhost:9000" setClientNamenodeAddress(conf); String nsId = getNameServiceId(conf); String namenodeId = HAUtil.getNameNodeId(conf, nsId); // HA相关 this.haEnabled = HAUtil.isHAEnabled(conf, nsId); state = createHAState(getStartupOption(conf)); this.allowStaleStandbyReads = HAUtil.shouldAllowStandbyReads(conf); this.haContext = createHAContext(); try { initializeGenericKeys(conf, nsId, namenodeId); // 完成实际的初始化工作 initialize(conf); // HA相关 try { haContext.writeLock(); state.prepareToEnterState(haContext); state.enterState(haContext); } finally { haContext.writeUnlock(); } } catch (IOException e) { this.stop(); throw e; } catch (HadoopIllegalArgumentException e) { this.stop(); throw e; } } //尽管本地没有开启HA（haEnabled=false**），**namenode依然拥有一个HAState，namenode的HAState状态为active. ---------------------------------------------------------------------- // 完成实际的初始化工作 // initialize(conf); protected void initialize(Configuration conf) throws IOException { if (conf.get(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS) == null) { String intervals = conf.get(DFS_METRICS_PERCENTILES_INTERVALS_KEY); if (intervals != null) { conf.set(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS, intervals); } } UserGroupInformation.setConfiguration(conf); loginAsNameNodeUser(conf); // 初始化metric NameNode.initMetrics(conf, this.getRole()); StartupProgressMetrics.register(startupProgress); // 启动httpServer if (NamenodeRole.NAMENODE == role) { startHttpServer(conf); } this.spanReceiverHost = SpanReceiverHost.getInstance(conf); // 从namenode目录加载fsimage与editlog，初始化FsNamesystem、FsDirectory、LeaseManager等 loadNamesystem(conf); // 创建RpcServer，封装了NameNodeRpcServer clientRpcServer，支持ClientNamenodeProtocol、DatanodeProtocolPB等协议 rpcServer = createRpcServer(conf); if (clientNamenodeAddress == null) { // This is expected for MiniDFSCluster. Set it now using // the RPC server's bind address. clientNamenodeAddress = NetUtils.getHostPortString(rpcServer.getRpcAddress()); LOG.info("Clients are to use " + clientNamenodeAddress + " to access" + " this namenode/service."); } if (NamenodeRole.NAMENODE == role) { httpServer.setNameNodeAddress(getNameNodeAddress()); httpServer.setFSImage(getFSImage()); } // 启动JvmPauseMonitor等，反向监控JVM pauseMonitor = new JvmPauseMonitor(conf); pauseMonitor.start(); metrics.getJvmMetrics().setPauseMonitor(pauseMonitor); // 启动执行多个非常重要工作的多个线程 startCommonServices(conf); } ---------------------------------------------------------------------- private void startCommonServices(Configuration conf) throws IOException { // 创建NameNodeResourceChecker、激活BlockManager等 namesystem.startCommonServices(conf, haContext); registerNNSMXBean(); // 角色非`NamenodeRole.NAMENODE`的在此处启动HttpServer if (NamenodeRole.NAMENODE != role) { startHttpServer(conf); httpServer.setNameNodeAddress(getNameNodeAddress()); httpServer.setFSImage(getFSImage()); } // 启动RPCServer rpcServer.start(); ...// 启动各插件 LOG.info(getRole() + " RPC up at: " + rpcServer.getRpcAddress()); if (rpcServer.getServiceRpcAddress() != null) { LOG.info(getRole() + " service RPC up at: " + rpcServer.getServiceRpcAddress()); } } -------------------------------------------------------------------------------------- void startCommonServices(Configuration conf, HAContext haContext) throws IOException { this.registerMBean(); // register the MBean for the FSNamesystemState writeLock(); this.haContext = haContext; try { // 创建NameNodeResourceChecker，并立即检查一次 nnResourceChecker = new NameNodeResourceChecker(conf); checkAvailableResources(); assert safeMode != null && !isPopulatingReplQueues(); // 设置一些启动过程中的信息 StartupProgress prog = NameNode.getStartupProgress(); prog.beginPhase(Phase.SAFEMODE); prog.setTotal(Phase.SAFEMODE, STEP_AWAITING_REPORTED_BLOCKS, getCompleteBlocksTotal()); // 设置已完成的数据块总量 setBlockTotal(); // 激活BlockManager blockManager.activate(conf); } finally { writeUnlock(); } registerMXBean(); DefaultMetricsSystem.instance().register(this); snapshotManager.registerMXBean(); } //blockManager.activate(conf)激活BlockManager主要完成PendingReplicationMonitor、DecommissionManager#Monitor、HeartbeatManager#Monitor、ReplicationMonitor public void activate(Configuration conf) { // 启动PendingReplicationMonitor pendingReplications.start(); // 激活DatanodeManager：启动DecommissionManager--Monitor、HeartbeatManager--Monitor datanodeManager.activate(conf); // 启动BlockManager--ReplicationMonitor this.replicationThread.start(); }namenode的主要责任是文件元信息与数据块映射的管理。相应的，namenode的启动流程需要关注与客户端、datanode通信的工作线程，文件元信息的管理机制，数据块的管理机制等。其中，RpcServer主要负责与客户端、datanode通信，FSDirectory主要负责管理文件元信息。DataNode 启动流程datanode的Main Class是DataNode，先找到DataNode.main()public class DataNode extends ReconfigurableBase implements InterDatanodeProtocol, ClientDatanodeProtocol, TraceAdminProtocol, DataNodeMXBean, ReconfigurationProtocol { public static final Logger LOG = LoggerFactory.getLogger(DataNode.class); static{ HdfsConfiguration.init(); } public static void main(String args[]) { if (DFSUtil.parseHelpArgument(args, DataNode.USAGE, System.out, true)) { System.exit(0); } secureMain(args, null); } ... public static void secureMain(String args[], SecureResources resources) { int errorCode = 0; try { // 打印启动信息 StringUtils.startupShutdownMessage(DataNode.class, args, LOG); // 完成创建datanode的主要工作 DataNode datanode = createDataNode(args, null, resources); if (datanode != null) { datanode.join(); } else { errorCode = 1; } } catch (Throwable e) { LOG.fatal("Exception in secureMain", e); terminate(1, e); } finally { LOG.warn("Exiting Datanode"); terminate(errorCode); } } --------------------------------------------------- public static DataNode createDataNode(String args[], Configuration conf, SecureResources resources) throws IOException { // 完成大部分初始化的工作，并启动部分工作线程 DataNode dn = instantiateDataNode(args, conf, resources); if (dn != null) { // 启动剩余工作线程 dn.runDatanodeDaemon(); } return dn; } --------------------------------------------------If this thread is specifically interrupted, it will stop waiting.*/ public void runDatanodeDaemon() throws IOException { // 在DataNode.instantiateDataNode()执行过程中会调用该方法（见后） blockPoolManager.startAll(); dataXceiverServer.start(); if (localDataXceiverServer != null) { localDataXceiverServer.start(); } ipcServer.start(); startPlugins(conf);}public static DataNode instantiateDataNode(String args [], Configuration conf, SecureResources resources) throws IOException { if (conf == null) conf = new HdfsConfiguration(); ... // 参数检查等 Collection dataLocations = getStorageLocations(conf); UserGroupInformation.setConfiguration(conf); SecurityUtil.login(conf, DFS_DATANODE_KEYTAB_FILE_KEY, DFS_DATANODE_KERBEROS_PRINCIPAL_KEY); return makeInstance(dataLocations, conf, resources);}//DataNode.makeInstance()开始创建DataNode static DataNode makeInstance(Collection dataDirs, Configuration conf, SecureResources resources) throws IOException { ...// 检查数据目录的权限 assert locations.size() > 0 : "number of data directories should be > 0"; return new DataNode(conf, locations, resources); } ... DataNode(final Configuration conf, final List dataDirs, final SecureResources resources) throws IOException { super(conf); ...// 参数设置 try { hostName = getHostName(conf); LOG.info("Configured hostname is " + hostName); startDataNode(conf, dataDirs, resources); } catch (IOException ie) { shutdown(); throw ie; } } ... void startDataNode(Configuration conf, List dataDirs, SecureResources resources) throws IOException { ...// 参数设置 // 初始化DataStorage storage = new DataStorage(); // global DN settings // 注册JMX registerMXBean(); // 初始化DataXceiver（流式通信），DataNode runDatanodeDaemon()中启动 initDataXceiver(conf); // 启动InfoServer（Web UI） startInfoServer(conf); // 启动JVMPauseMonitor（反向监控JVM情况，可通过JMX查询） pauseMonitor = new JvmPauseMonitor(conf); pauseMonitor.start(); ...// 略 // 初始化IpcServer（RPC通信），DataNode-runDatanodeDaemon()中启动 initIpcServer(conf); metrics = DataNodeMetrics.create(conf, getDisplayName()); metrics.getJvmMetrics().setPauseMonitor(pauseMonitor); // 按照namespace（nameservice）、namenode的结构进行初始化 blockPoolManager = new BlockPoolManager(this); blockPoolManager.refreshNamenodes(conf); ...// 略 } //BlockPoolManager抽象了datanode提供的数据块存储服务。BlockPoolManager按照namespace（nameservice）、namenode结构组织。 //BlockPoolManager-refreshNamenodes() //除了初始化过程主动调用，还可以由namespace通过datanode心跳过程下达刷新命令 void refreshNamenodes(Configuration conf) throws IOException { LOG.info("Refresh request received for nameservices: " + conf.get(DFSConfigKeys.DFS_NAMESERVICES)); Map<String, Map<String, InetSocketAddress>> newAddressMap = DFSUtil.getNNServiceRpcAddressesForCluster(conf); synchronized (refreshNamenodesLock) { doRefreshNamenodes(newAddressMap); }}private void doRefreshNamenodes( Map<String, Map<String, InetSocketAddress>> addrMap) throws IOException { assert Thread.holdsLock(refreshNamenodesLock); Set toRefresh = Sets.newLinkedHashSet(); Set toAdd = Sets.newLinkedHashSet(); Set toRemove; synchronized (this) { // Step 1. For each of the new nameservices, figure out whether // it's an update of the set of NNs for an existing NS, // or an entirely new nameservice. for (String nameserviceId : addrMap.keySet()) { if (bpByNameserviceId.containsKey(nameserviceId)) { toRefresh.add(nameserviceId); } else { toAdd.add(nameserviceId); }} ...// 略 // Step 2. Start new nameservices if (!toAdd.isEmpty()) { LOG.info("Starting BPOfferServices for nameservices: " + Joiner.on(",").useForNull("<default>").join(toAdd)); for (String nsToAdd : toAdd) { ArrayList<InetSocketAddress> addrs = Lists.newArrayList(addrMap.get(nsToAdd).values()); // 为每个namespace创建对应的BPOfferService BPOfferService bpos = createBPOS(addrs); bpByNameserviceId.put(nsToAdd, bpos); offerServices.add(bpos); }} // 然后通过startAll启动所有BPOfferService startAll(); } ...// 略}protected BPOfferService createBPOS(List nnAddrs) { return new BPOfferService(nnAddrs, dn); } BPOfferService(List nnAddrs, DataNode dn) { Preconditions.checkArgument(!nnAddrs.isEmpty(), "Must pass at least one NN."); this.dn = dn; for (InetSocketAddress addr : nnAddrs) { this.bpServices.add(new BPServiceActor(addr, this)); }}//BlockPoolManager#startAll()启动所有BPOfferService（实际是启动所有BPServiceActor）。 synchronized void startAll() throws IOException { try { UserGroupInformation.getLoginUser().doAs( new PrivilegedExceptionAction() { @Override public Object run() throws Exception { for (BPOfferService bpos : offerServices) { bpos.start(); } return null; }} ); } catch (InterruptedException ex) { IOException ioe = new IOException(); ioe.initCause(ex.getCause()); throw ioe; }}//在datanode启动的主流程中，启动了多种工作线程，包括InfoServer、JVMPauseMonitor、BPServiceActor等。其中，最重要的是BPServiceActor线程，真正代表datanode与namenode通信的正是BPServiceActor线程。 //DataNode--initBlockPool()： /**One of the Block Pools has successfully connected to its NN.This initializes the local storage for that block pool,checks consistency of the NN's cluster ID, etc.*If this is the first block pool to register, this also initializesthe datanode-scoped storage.*@param bpos Block pool offer service@throws IOException if the NN is inconsistent with the local storage.*/ void initBlockPool(BPOfferService bpos) throws IOException { ...// 略 // 将blockpool注册到BlockPoolManager blockPoolManager.addBlockPool(bpos); // 初步初始化存储结构 initStorage(nsInfo); ...// 检查磁盘损坏 // 启动扫描器 initPeriodicScanners(conf); // 将blockpool添加到FsDatasetIpml，并继续初始化存储结构 data.addBlockPool(nsInfo.getBlockPoolID(), conf); }# NameNode如何支撑高并发访问（双缓冲机制） 高并发访问NameNode会遇到什么样的问题：写入本地磁盘--edits文件通过网络传输给JournalNodes集群（Hadoop HA集群--结合zookeeper来学习）。高并发的难点主要在于数据的多线程安全以及每个操作效率。对于多线程安全：NameNode在写edits log时几个原则：写入数据到edits_log必须保证每条edits都有一个全局顺序递增的transactionId（简称为txid），这样才可以标识出来一条一条的edits的先后顺序。如果要保证每条edits的txid都是递增的，就必须得加同步锁。也就是每个线程修改了元数据，要写一条edits 的时候，都必须按顺序排队获取锁后，才能生成一个递增的txid，代表这次要写的edits的序号。产生的问题：如果每次都是在一个加锁的代码块里，生成txid，然后写磁盘文件edits log，这种既有同步锁又有写磁盘操作非常耗时。HDFS优化解决方案问题产生的原因主要是在于，写edits时串行化排队生成自增txid + 写磁盘操作费时，HDFS的解决方案串行化：使用分段锁写磁盘：使用双缓冲分段加锁机制首先各个线程依次第一次获取锁，生成顺序递增的txid，然后将edits写入内存双缓冲的区域1，接着就立马第一次释放锁了。趁着这个空隙，后面的线程就可以再次立马第一次获取锁，然后立即写自己的edits到内存缓冲。双缓冲机制程序中将会开辟两份一模一样的内存空间，一个为bufCurrent，产生的数据会直接写入到这个bufCurrent，而另一个叫bufReady，在bufCurrent数据写入(达到一定标准)后，两片内存就会exchange（交换）。直接交换双缓冲的区域1和区域2。保证接收客户端写入数据请求的都是操作内存而不是同步写磁盘。双缓冲源码分析 找到FsEditLog.java... void logEdit(final FSEditLogOp op) { Boolean needsSync = false; //是否同步的标识 synchronized (this) { // assert isOpenForWrite() : "bad state: " + state; // wait if an automatic sync is scheduled 如果当前操作被其它线程调度，则等待1s钟 waitIfAutoSyncScheduled(); // check if it is time to schedule an automatic sync needsSync = doEditTransaction(op); if (needsSync) { isAutoSyncScheduled = true; //标识bufCurrent满了，进行双缓冲刷写 } } // Sync the log if an automatic sync is required. if (needsSync) { logSync(); //将缓冲区数据刷写到磁盘 } } ...
Yarn架构ResourceManager（rm）：处理客户端请求、启动/监控ApplicationMaster、监控NodeManager、资源分配与调度。NodeManager（nm）：单个节点上的资源管理、处理来自ResourceManager的命令、处理来自ApplicationMaster的命令。ApplicationMaster（am）：数据切分、为应用程序申请资源，并分配给内部任务、任务监控与容错。Container：对任务运行环境的抽象，封装了CPU、内存等多维资源以及环境变量、启动命令等任务运行相关的信息。Yarn任务提交（工作机制）作业提交过程之Yarn：作业提交第1步：Client调用job.waitForCompletion方法，向整个集群提交MapReduce作业。第2步：Client向RM申请一个作业id。第3步：RM给Client返回该job资源的提交路径和作业id。第4步：Client提交jar包、切片信息和配置文件到指定的资源提交路径。第5步：Client提交完资源后，向RM申请运行MrAppMaster。作业初始化第6步：当RM收到Client的请求后，将该job添加到容量调度器中。第7步：某一个空闲的NM领取到该Job。第8步：该NM创建Container，并产生MRAppmaster。第9步：下载Client提交的资源到本地。任务分配第10步：MrAppMaster向RM申请运行多个MapTask任务资源。第11步：RM将运行MapTask任务分配给另外两个NodeManager，另两个NodeManager分别领取任务并创建容器。任务运行第12步：MR向两个接收到任务的NodeManager发送程序启动脚本，这两个NodeManager分别启动MapTask，MapTask对数据分区排序。第13步：MrAppMaster等待所有MapTask运行完毕后，向RM申请容器，运行ReduceTask。第14步：ReduceTask向MapTask获取相应分区的数据。第15步：程序运行完毕后，MR会向RM申请注销自己。进度和状态更新YARN中的任务将其进度和状态返回给应用管理器，客户端每秒(通过mapreduce.client.progressmonitor.pollinterval设置)向应用管理器请求进度更新，展示给用户。作业完成除了向应用管理器请求作业进度外，客户端每5秒都会通过调用waitForCompletion()来检查作业是否完成。时间间隔可以通过mapreduce.client.completion.pollinterval来设置。作业完成之后，应用管理器和Container会清理工作状态。作业的信息会被作业历史服务器存储以备之后用户核查。Yarn调度策略Hadoop作业调度器主要有三种：FIFO、Capacity Scheduler和Fair Scheduler。Hadoop2.9.2默认的资源调度器是Capacity Scheduler。可以查看yarn-default.xmlFIFO（先进先出调度器）容量调度器（Capacity Scheduler 默认的调度器）Apache Hadoop默认使用的调度策略。Capacity 调度器允许多个组织共享整个集群，每个组织可以获得集群的一部分计算能力。通过为每个组织分配专门的队列，然后再为每个队列分配一定的集群资源，这样整个集群就可以通过设置多个队列的方式给多个组织提供服务了。除此之外，队列内部又可以垂直划分，这样一个组织内部的多个成员就可以共享这个队列资源了，在一个队列内部，资源的调度是采用的是先进先出(FIFO)策略。Fair Scheduler（公平调度器，CDH版本的hadoop默认使用的调度器）Fair调度器的设计目标是为所有的应用分配公平的资源（对公平的定义可以通过参数来设置）。公平调度在也可以在多个队列间工作。举个例子，假设有两个用户A和B，他们分别拥有一个队列。当A启动一个job而B没有任务时，A会获得全部集群资源；当B启动一个job后，A的job会继续运行，不过一会儿之后两个任务会各自获得一半的集群资源。如果此时B再启动第二个job并且其它job还在运行，则它将会和B的第一个job共享B这个队列的资源，也就是B的两个job会用于四分之一的集群资源，而A的job仍然用于集群一半的资源，结果就是资源最终在两个用户之间平等的共享。Yarn多租户资源隔离配置Yarn集群资源设置为A，B两个队列，A队列设置占用资源70%主要用来运行常规的定时任务，B队列设置占用资源30%主要运行临时任务，两个队列间可相互资源共享，假如A队列资源占满，B队列资源比较充裕，A队列可以使用B队列的资源，使总体做到资源利用最大化。选择使用Fair Scheduler调度策略：具体配置yarn-site.xml <property> <name>yarn.resourcemanager.scheduler.class</name> <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value> <description>In case you do not want to use the defaultscheduler</description> </property>创建fair-scheduler.xml文件在Hadoop安装目录/etc/hadoop创建该文件<?xml version="1.0" encoding="UTF-8" standalone="yes" ?> <allocations> <defaultQueueSchedulingPolicy>fair</defaultQueueSchedulingPolicy> <queue name="root"> <queue name="default"> <aclAdministerApps>*</aclAdministerApps> <aclSubmitApps>*</aclSubmitApps> <maxResources>9216 mb,4 vcores</maxResources> <maxRunningApps>100</maxRunningApps> <minResources>1024 mb,1vcores</minResources> <minSharePreemptionTimeout>1000</minSharePreemptionTimeout> <schedulingPolicy>fair</schedulingPolicy> <weight>7</weight> </queue> <queue name="queue1"> <aclAdministerApps>*</aclAdministerApps> <aclSubmitApps>*</aclSubmitApps> <maxResources>4096 mb,4vcores</maxResources> <maxRunningApps>5</maxRunningApps> <minResources>1024 mb, 1vcores</minResources> <minSharePreemptionTimeout>1000</minSharePreemptionTimeout> <schedulingPolicy>fair</schedulingPolicy> <weight>3</weight> </queue> </queue> <queuePlacementPolicy> <rule create="false" name="specified" /> <rule create="true" name="default" /> </queuePlacementPolicy> </allocations>界面验证
MapReduce思想MapReduce思想在生活中处处可见。我们或多或少都曾接触过这种思想。MapReduce的思想核心是分而治之，充分利用了并行处理的优势。即使是发布过论文实现分布式计算的谷歌也只是实现了这种思想，而不是自己原创。MapReduce任务过程是分为两个处理阶段：Map阶段：Map阶段的主要作用是“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。Map阶段的这些任务可以并行计算，彼此间没有依赖关系。Reduce阶段：Reduce阶段的主要作用是“合”，即对map阶段的结果进行全局汇总。再次理解MapReduce的思想官方WordCount案例源码解析经过查看分析官方WordCount案例源码我们发现一个统计单词数量的MapReduce程序的代码由三个部分组成：Mapper类Reducer类运行作业的代码（Driver）Mapper类继承了org.apache.hadoop.mapreduce.Mapper类重写了其中的map方法，Reducer类继承了org.apache.hadoop.mapreduce.Reducer类重写了其中的reduce方法。重写的Map方法作用：map方法其中的逻辑就是用户希望mr程序map阶段如何处理的逻辑；重写的Reduce方法作用：reduce方法其中的逻辑是用户希望mr程序reduce阶段如何处理的逻辑；Hadoop序列化为什么进行序列化？序列化主要是我们通过网络通信传输数据时或者把对象持久化到文件，需要把对象序列化成二进制的结构。观察源码时发现自定义Mapper类与自定义Reducer类都有泛型类型约束，比如自定义Mapper有四个形参类型，但是形参类型并不是常见的java基本类型。为什么Hadoop要选择建立自己的序列化格式而不使用java自带serializable？序列化在分布式程序中非常重要，在Hadoop中，集群中多个节点的进程间的通信是通过RPC（远程过程调用：Remote Procedure Call）实现；RPC将消息序列化成二进制流发送到远程节点，远程节点再将接收到的二进制数据反序列化为原始的消息，因此RPC往往追求如下特点：紧凑:数据更紧凑，能充分利用网络带宽资源快速:序列化和反序列化的性能开销更低Hadoop使用的是自己的序列化格式Writable,它比java的序列化serialization更紧凑速度更快。一个对象使用Serializable序列化后，会携带很多额外信息比如校验信息，Header，继承体系等。Java基本类型与Hadoop常用序列化类型Java基本类型Hadoop Writable类型booleanBooleanWritablebyteByteWritableintIntWritablefloatFloatWritablelongLongWritabledoubleDoubleWritableStringTextmapMapWritablearrayArrayWritableMapReduce编程规范及示例编写Mapper类用户自定义一个Mapper类继承Hadoop的Mapper类Mapper的输入数据是KV对的形式（类型可以自定义）Map阶段的业务逻辑定义在map()方法中Mapper的输出数据是KV对的形式（类型可以自定义）注意：map()方法是对输入的一个KV对调用一次！！Reducer类用户自定义Reducer类要继承Hadoop的Reducer类Reducer的输入数据类型对应Mapper的输出数据类型（KV对）Reducer的业务逻辑写在reduce()方法中Reduce()方法是对相同K的一组KV对调用执行一次Driver阶段创建提交YARN集群运行的Job对象，其中封装了MapReduce程序运行所需要的相关参数入输入数据路径，输出数据路径等，也相当于是一个YARN集群的客户端，主要作用就是提交我们MapReduce程序运行。WordCount代码实现需求在给定的文本文件中统计输出每一个单词出现的总次数输入数据：wc.txt;输出：apache 2clickhouse 2hadoop 1mapreduce 1spark 2xiaoming 1具体步骤按照MapReduce编程规范，分别编写Mapper，Reducer，Driver。新建maven工程导入hadoop依赖<dependencies> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.9.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> 1.8 <target>1.8</target> </configuration> </plugin> <plugin> <artifactId>maven-assembly-plugin </artifactId> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build>添加log4j.propertieslog4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n整体思路梳理（仿照源码）Map阶段：map()方法中把传入的数据转为String类型根据空格切分出单词输出<单词，1>Reduce阶段：汇总各个key(单词)的个数，遍历value数据进行累加输出key的总数Driver获取配置文件对象，获取job对象实例指定程序jar的本地路径指定Mapper/Reducer类指定Mapper输出的kv数据类型指定最终输出的kv数据类型指定job处理的原始数据路径指定job输出结果路径提交作业编写Mapper类import java.io.IOException; 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 WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ Text k = new Text(); IntWritable v = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1 获取一行 String line = value.toString(); // 2 切割 String[] words = line.split(" "); // 3 输出 for (String word : words) { k.set(word); context.write(k, v); } } }继承的Mapper类型选择新版本API：编写Reducer类import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ int sum; IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException { // 1 累加求和 sum = 0; for (IntWritable count : values) { sum += count.get(); } // 2 输出 v.set(sum); context.write(key,v); } }选择继承的Reducer类：编写Driver驱动类import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordcountDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1 获取配置信息以及封装任务 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 设置jar加载路径 job.setJarByClass(WordcountDriver.class); // 3 设置map和reduce类 job.setMapperClass(WordcountMapper.class); job.setReducerClass(WordcountReducer.class); // 4 设置map输出 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5 设置最终输出kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6 设置输入和输出路径 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 提交 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }运行任务本地模式直接Idea中运行驱动类即可idea运行需要传入参数：在program arguments设置参数运行结束，去到输出结果路径查看结果注意本地idea运行mr任务与集群没有任何关系，没有提交任务到yarn集群，是在本地使用多线程方式模拟的mr的运行。Yarn集群模式1）把程序打成jar包，改名为wc.jar;上传到Hadoop集群选择合适的Jar包准备原始数据文件，上传到HDFS的路径，不能是本地路径，因为跨节点运行无法获取数据！！2）启动Hadoop集群（Hdfs,Yarn）3）使用Hadoop 命令提交任务运行hadoop jar wc.jar com.test.wordcount.WordcountDriver /user/test/input /user/test/outputYarn集群任务运行成功展示图序列化Writable接口基本序列化类型往往不能满足所有需求，比如在Hadoop框架内部传递一个自定义bean对象，那么该对象就需要实现Writable序列化接口。实现Writable序列化步骤如下必须实现Writable接口反序列化时，需要反射调用空参构造函数，所以必须有空参构造public CustomBean() { super(); }重写序列化方法@Override public void write(DataOutput out) throws IOException { .... }重写反序列化方法@Override public void readFields(DataInput in) throws IOException { .... }反序列化的字段顺序和序列化字段的顺序必须完全一致方便展示结果数据，需要重写bean对象的toString()方法，可以自定义分隔符如果自定义Bean对象需要放在Mapper输出KV中的K,则该对象还需实现Comparable接口，因为因为MapReduce框中的Shuffle过程要求对key必须能排序！！排序内容后面说！@Override public int compareTo(CustomBean o) { // 自定义排序规则 return this.num > o.getNum() ? -1 : 1; }Writable接口案例需求统计每台智能音箱设备内容播放时长原始日志格式：001577c3 kar_890809 120.196.100.99 1116 954 200日志id 设备id appkey(合作硬件厂商) 网络ip 自有内容时长(秒) 第三方内容时长(秒) 网络状态码输出结果：001577c3 11160 9540 20700设备id 自有内容时长(秒) 第三方内容时长(秒) 总时长编写MapReduce程序创建SpeakBean对象package com.test.hdfs; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; // 1 实现writable接口 public class SpeakBean implements Writable { private long selfDuration; private long thirdPartDuration; private long sumDuration; //2 反序列化时，需要反射调用空参构造函数，所以必须有 public SpeakBean() { } public SpeakBean(long selfDuration, long thirdPartDuration) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.sumDuration=this.selfDuration+this.thirdPartDuration; } //3 写序列化方法 public void write(DataOutput out) throws IOException { out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeLong(sumDuration); } //4 反序列化方法 //5 反序列化方法读顺序必须和写序列化方法的写顺序必须一致 public void readFields(DataInput in) throws IOException { this.selfDuration = in.readLong(); this.thirdPartDuration = in.readLong(); this.sumDuration = in.readLong(); } // 6 编写toString方法，方便后续打印到文本 @Override public String toString() { return selfDuration + "\t" + thirdPartDuration + "\t" + sumDuration ; } public long getSelfDuration() { return selfDuration; } public void setSelfDuration(long selfDuration) { this.selfDuration = selfDuration; } public long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public long getSumDuration() { return sumDuration; } public void setSumDuration(long sumDuration) { this.sumDuration = sumDuration; } public void set(long selfDuration, long thirdPartDuration) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.sumDuration=this.selfDuration+this.thirdPartDuration; } }编写Mapper类package com.test.hdfs; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SpeakDurationMapper extends Mapper<LongWritable, Text, Text, SpeakBean> { SpeakBean v = new SpeakBean(); Text k = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1 获取一行 String line = value.toString(); // 2 切割字段 String[] fields = line.split("\t"); // 3 封装对象 // 取出设备id String deviceId = fields[1]; // 取出自有和第三方时长数据 long selfDuration = Long.parseLong(fields[fields.length - 3]); long thirdPartDuration = Long.parseLong(fields[fields.length - 2]); k.set(deviceId); v.set(selfDuration, thirdPartDuration); // 4 写出 context.write(k, v); } }编写Reducerpackage com.test.hdfs; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SpeakDurationReducer extends Reducer<Text, SpeakBean, Text, SpeakBean> { @Override protected void reduce(Text key, Iterable<SpeakBean> values, Context context)throws IOException, InterruptedException { long self_Duration = 0; long thirdPart_Duration = 0; // 1 遍历所用bean，将其中的自有，第三方时长分别累加 for (SpeakBean sb : values) { self_Duration += sb.getSelfDuration(); thirdPart_Duration += sb.getThirdPartDuration(); } // 2 封装对象 SpeakBean resultBean = new SpeakBean(self_Duration, thirdPart_Duration); // 3 写出 context.write(key, resultBean); } }编写驱动package com.test.hdfs; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class SpeakerDriver { public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException { // 输入输出路径需要根据自己电脑上实际的输入输出路径设置 args = new String[] { "e:/input/input", "e:/output1" }; // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 6 指定本程序的jar包所在的本地路径 job.setJarByClass(SpeakerDriver.class); // 2 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(SpeakDurationMapper.class); job.setReducerClass(SpeakDurationReducer.class); // 3 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(SpeakBean.class); // 4 指定最终输出的数据的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(SpeakBean.class); // 5 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }mr编程技巧总结结合业务设计Map输出的key和v，利用key相同则去往同一个reduce的特点！！map()方法中获取到只是一行文本数据尽量不做聚合运算reduce()方法的参数要清楚含义MapReduce原理分析MapTask运行机制详解MapTask流程详细步骤：首先，读取数据组件InputFormat（默认TextInputFormat）会通过getSplits方法对输入目录中文件进行逻辑切片规划得到splits，有多少个split就对应启动多少个MapTask。split与block的对应关系默认是一对一。将输入文件切分为splits之后，由RecordReader对象（默认LineRecordReader）进行读取，以\n作为分隔符，读取一行数据，返回<key，value>。Key表示每行首字符偏移值，value表示这一行文本内容。读取split返回<key,value>，进入用户自己继承的Mapper类中，执行用户重写的map函数。RecordReader读取一行这里调用一次。map逻辑完之后，将map的每条结果通过context.write进行collect数据收集。在collect中，会先对其进行分区处理，默认使用HashPartitioner。MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。接下来，会将数据写入内存，内存中这片区域叫做环形缓冲区，缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。1）环形缓冲区其实是一个数组，数组中存放着key、value的序列化数据和key、value的元数据信息，包括partition、key的起始位置、value的起始位置以及value的长度。环形结构是一个抽象概念。2）缓冲区是有大小限制，默认是100MB。当map task的输出结果很多时，就可能会撑爆内存，所以需要在一定条件下将缓冲区中的数据临时写入磁盘，然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill，中文可译为溢写。这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8，也就是当缓冲区的数据已经达到阈值（buffer size spill percent = 100MB 0.8 = 80MB），溢写线程启动，锁定这80MB的内存，执行溢写过程。Maptask的输出结果还可以往剩下的20MB内存中写，互不影响。当溢写线程启动后，需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为!1）如果job设置过Combiner，那么现在就是使用Combiner的时候了。将有相同key的key/value对的value加起来，减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果，所以它在整个模型中会多次使用。2）那哪些场景才能使用Combiner呢？从这里分析，Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。Combiner的使用一定得慎重，如果用好，它对job执行效率有帮助，反之会影响reduce的最终结果。合并溢写文件：每次溢写会在磁盘上生成一个临时文件（写之前判断是否有combiner），如果map的输出结果真的很大，有多次这样的溢写发生，磁盘上相应的就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行merge合并，因为最终的文件只有一个，写入磁盘，并且为这个文件提供了一个索引文件，以记录每个reduce对应数据的偏移量。至此map整个阶段结束!!MapTask的一些配置：官方参考地址：https://hadoop.apache.org/docs/r2.9.2/hadoop-mapreduce-client/hadoop-mapreduceclient-core/mapred-default.xmlMapTask的并行度MapTask并行度思考MapTask的并行度决定Map阶段的任务处理并发度，从而影响到整个Job的处理速度。思考：MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？MapTask并行度决定机制数据块：Block是HDFS物理上把数据分成一块一块。切片：数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。切片机制源码阅读默认就是128M；MapTask并行度是不是越多越好呢？答案不是，如果一个文件仅仅比128M大一点点也被当成一个split来对待，而不是多个split。MR框架在并行运算的同时也会消耗更多资源，并行度越高资源消耗也越高，假设129M文件分为两个分片，一个是128M，一个是1M；对于1M的切片的Maptask来说，太浪费资源。思考：129M的文件在Hdfs存储的时候会不会切成两块？ReduceTask 工作机制Reduce大致分为copy、sort、reduce三个阶段，重点在前两个阶段。copy阶段包含一个eventFetcher来获取已完成的map列表，由Fetcher线程去copy数据，在此过程中会启动两个merge线程，分别为inMemoryMerger和onDiskMerger，分别将内存中的数据merge到磁盘和将磁盘中的数据进行merge。待数据copy完成之后，copy阶段就完成了，开始进行sort阶段，sort阶段主要是执行finalMerge操作，纯粹的sort阶段，完成之后就是reduce阶段，调用用户定义的reduce函数进行处理。详细步骤：Copy阶段，简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher)，通过HTTP方式请求maptask获取属于自己的文件。Merge阶段。这里的merge如map端的merge动作，只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活。merge有三种形式：内存到内存；内存到磁盘；磁盘到磁盘。默认情况下第一种形式不启用。当内存中的数据量到达一定阈值，就启动内存到磁盘的merge。与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行，直到没有map端的数据时才结束，然后启动第三种磁盘到磁盘的merge方式生成最终的文件。合并排序。把分散的数据合并成一个大的数据后，还会再对合并后的数据排序。对排序后的键值对调用reduce方法，键相等的键值对调用一次reduce方法，每次调用会产生零个或者多个键值对，最后把这些输出的键值对写入到HDFS文件中。ReduceTask并行度ReduceTask的并行度同样影响整个Job的执行并发度和执行效率，但与MapTask的并发数由切片数决定不同，ReduceTask数量的决定是可以直接手动设置：// 默认值是1，手动设置为4 job.setNumReduceTasks(4);注意事项ReduceTask=0，表示没有Reduce阶段，输出文件数和MapTask数量保持一致；ReduceTask数量不设置默认就是一个，输出文件数量为1个；如果数据分布不均匀，可能在Reduce阶段产生倾斜；Shuffle机制map阶段处理的数据如何传递给reduce阶段，是MapReduce框架中最关键的一个流程，这个流程就叫shuffle。shuffle: 洗牌、发牌——（核心机制：数据分区，排序，分组，combine，合并等过程）MapReduce的分区与reduceTask的数量在MapReduce中，通过我们指定分区，会将同一个分区的数据发送到同一个reduce当中进行处理(默认是key相同去往同个分区)，例如我们为了数据的统计，我们可以把一批类似的数据发送到同一个reduce当中去，在同一个reduce当中统计相同类型的数据，如何才能保证相同key的数据去往同个reduce呢？只需要保证相同key的数据分发到同个分区即可。结合以上原理分析我们知道MR程序shuffle机制默认就是这种规则！！分区源码翻阅源码验证以上规则，MR程序默认使用的HashPartitioner，保证了相同的key去往同个分区！！自定义分区实际生产中需求变化多端，默认分区规则往往不能满足需求，需要结合业务逻辑来灵活控制分区规则以及分区数量！！如何制定自己需要的分区规则？具体步骤：1）自定义类继承Partitioner，重写getPartition()方法2）在Driver驱动中，指定使用自定义Partitioner3）在Driver驱动中，要根据自定义Partitioner的逻辑设置相应数量的ReduceTask数量。案例：需求：按照不同的appkey把记录输出到不同的分区中原始日志格式：001577c3 kar_890809 120.196.100.99 1116 954 200日志id 设备id appkey(合作硬件厂商) 网络ip 自有内容时长(秒) 第三方内容时长(秒) 网络状态码输出结果：根据appkey把不同厂商的日志数据分别输出到不同的文件中需求分析：面对业务需求，结合mr的特点，来设计map输出的kv,以及reduce输出的kv数据。一个ReduceTask对应一个输出文件，因为在shuffle机制中每个reduceTask拉取的都是某一个分区的数据，一个分区对应一个输出文件。结合appkey的前缀相同的特点，同时不能使用默认分区规则，而是使用自定义分区器，只要appkey前缀相同则数据进入同个分区。整体思路：Mapper：1）读取一行文本，按照制表符切分2）解析出appkey字段，其余数据封装为PartitionBean对象（实现序列化Writable接口）3）设计map()输出的kv,key-->appkey(依靠该字段完成分区)，PartitionBean对象作为Value输出Partition：自定义分区器，实现按照appkey字段的前缀来区分所属分区Reduce：reduce()正常输出即可，无需进行聚合操作Driver：1）在原先设置job属性的同时增加设置使用自定义分区器2）注意设置ReduceTask的数量（与分区数量保持一致）Mapperpackage com.test.mr.partition; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class PartitionMapper extends Mapper<LongWritable, Text, Text, PartitionBean> { PartitionBean bean = new PartitionBean(); Text text = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); String appkey = fields[2]; bean.setId(fields[0]); bean.setDeviceId(fields[1]); bean.setAppkey(appkey); bean.setIp(fields[3]); bean.setSelfDuration(Long.parseLong(fields[4])); bean.setThirdPartDuration(Long.parseLong(fields[5])); bean.setStatus(fields[6]); text.set(appkey); context.write(text, bean); } }PartitionBeanpackage com.test.mr.partition; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class PartitionBean implements Writable { //定义属性 private String id; //日志id private String deviceId;//设备id private String appkey;//appkey合作硬件厂商id private String ip;//ip地址 private Long selfDuration;//自有内容时长 private Long thirdPartDuration;//第三方内容时长 private String status;//状态码 public PartitionBean() { } public PartitionBean(String id, String deviceId, String appkey, String ip, Long selfDuration, Long thirdPartDuration, String status) { this.id = id; this.deviceId = deviceId; this.appkey = appkey; this.ip = ip; this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.status = status; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public String getAppkey() { return appkey; } public void setAppkey(String appkey) { this.appkey = appkey; } public String getIp() { return ip; } public void setIp(String ip) { this.ip = ip; } public Long getSelfDuration() { return selfDuration; } public void setSelfDuration(Long selfDuration) { this.selfDuration = selfDuration; } public Long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(Long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public String getStatus() { return status; } public void setStatus(String status) { this.status = status; } //序列化 @Override public void write(DataOutput out) throws IOException { out.writeUTF(id); out.writeUTF(deviceId); out.writeUTF(appkey); out.writeUTF(ip); out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeUTF(status); } //反序列化 @Override public void readFields(DataInput in) throws IOException { this.id = in.readUTF(); this.deviceId=in.readUTF(); this.appkey=in.readUTF(); this.ip=in.readUTF(); this.selfDuration=in.readLong(); this.thirdPartDuration=in.readLong(); this.status=in.readUTF(); } @Override public String toString() { return id + '\t' + deviceId + '\t' + appkey + '\t' + ip + '\t' + selfDuration +'\t'+ thirdPartDuration + '\t' + status ; } }CustomPartitionerpackage com.test.mr.partition; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class CustomPartitioner extends Partitioner<Text,PartitionBean> { @Override public int getPartition(Text text, PartitionBean partitionBean, int numPartitions) { int partition=0; String appkey = text.toString(); if(appkey.equals("kar")){ partition=1; }else if(appkey.equals("pandora")){ partition=2; }else{ partition=0; } return partition; } }PartitionReducerpackage com.test.mr.partition; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class PartitionReducer extends Reducer<Text, PartitionBean, NullWritable, PartitionBean> { @Override protected void reduce(Text key, Iterable<PartitionBean> values, Context context) throws IOException, InterruptedException { for (PartitionBean bean : values) { context.write(NullWritable.get(), bean); } } }PartitionDriverpackage com.test.mr.partition; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class PartitionDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(PartitionDriver.class); job.setMapperClass(PartitionMapper.class); job.setReducerClass(PartitionReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(PartitionBean.class); job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(PartitionBean.class); job.setPartitionerClass(CustomPartitioner.class); job.setNumReduceTasks(5); FileInputFormat.setInputPaths(job, new Path("e:/speak.data")); FileOutputFormat.setOutputPath(job, new Path("e:/partition/output")); final boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }总结1）自定义分区器时最好保证分区数量与reduceTask数量保持一致；2）如果分区数量不止1个，但是reduceTask数量1个，此时只会输出一个文件。3）如果reduceTask数量大于分区数量，但是输出多个空文件4）如果reduceTask数量小于分区数量，有可能会报错。MapReduce中的Combinercombiner运行机制：Combiner是MR程序中Mapper和Reducer之外的一种组件Combiner组件的父类就是ReducerCombiner和reducer的区别在于运行的位置Combiner是在每一个maptask所在的节点运行;Combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量。Combiner能够应用的前提是不能影响最终的业务逻辑，此外，Combiner的输出kv应该跟reducer的输入kv类型要对应起来。举例说明：假设一个计算平均值的MR任务Map阶段有2个MapTaskMapTask1输出数据：10,5,15 如果使用Combiner:(10+5+15)/3=10MapTask2输出数据：2,6 如果使用Combiner:(2+6)/2=4Reduce阶段汇总（10+4）/2=7而正确结果应该是（10+5+15+2+6）/5=7.6自定义Combiner实现步骤：自定义一个Combiner继承Reducer，重写Reduce方法在驱动(Driver)设置使用Combiner（默认是不适用Combiner组件）改造WordCount程序package com.test.mr.wc; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.io.Text; import java.io.IOException; public class WordCountCombiner extends Reducer<Text, IntWritable,Text,IntWritable> { IntWritable total = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { // 遍历key对应的values，然后累加结果 int sum = 0; for (IntWritable value : values) { int i = value.get(); sum += 1; } // 直接输出当前key对应的sum值，结果就是单词出现的总次数 total.set(sum); context.write(key, total); } }在驱动(Driver)设置使用Combinerjob.setCombinerClass(WordcountCombiner.class);验证结果，观察程序运行日志如果直接使用WordCountReducer作为Combiner使用是否可以？可以。MapReduce中的排序排序是MapReduce框架中最重要的操作之一。MapTask和ReduceTask均会对数据按照key进行排序。该操作属于Hadoop的默认行为。任何应用程序中的数据均会被排序，而不管逻辑上是否需要。默认排序是按照字典顺序排序，且实现该排序的方法是快速排序。MapTask它会将处理的结果暂时放到环形缓冲区中，当环形缓冲区使用率达到一定阈值后，再对缓冲区中的数据进行一次快速排序，并将这些有序数据溢写到磁盘上，溢写完毕后，它会对磁盘上所有文件进行归并排序。ReduceTask当所有数据拷贝完毕后，ReduceTask统-对内存和磁盘上的所有数据进行一次归并排序。部分排序.MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部有序。全排序最终输出结果只有一个文件，且文件内部有序。实现方式是只设置一个ReduceTask。但该方法在处理大型文件时效率极低，因为一台机器处理所有文件，完全丧失了MapReduce所提供的并行架构。辅助排序: (GroupingComparator分组)在Reduce端对key进行分组。应用于:在接收的key为bean对象时，想让一个或几个字段相同(全部字段比较不相同)的key进入到同一个reduce方法时，可以采用分组排序。4.二次排序.在自定义排序过程中，如果compareTo中的判断条件为两个即为二次排序。全排序WritableComparableBean对象如果作为Map输出的key时，需要实现WritableComparable接口并重写compareTo方法指定排序规则基于统计的播放时长案例的输出结果对总时长进行排序实现全局排序只能设置一个ReduceTask!!播放时长案例输出结果：00fdaf3 33180 33420 00fdaf3 6660000wersa4 30689 35191 00wersa4 658800a0fe2 43085 44254 0a0fe2 873390ad0s7 31702 29183 0ad0s7 608850sfs01 31883 29101 0sfs01 60984a00df6s 33239 36882 a00df6s 70121adfd00fd5 30727 31491 adfd00fd5 62218需求分析：如何设计map()方法输出的key和value？MR框架中shuffle阶段的排序是默认行为，不管你是否需要都会进行排序。key:把所有字段封装成为一个bean对象，并且指定bean对象作为key输出，如果作为key输出，需要实现排序接口，指定自己的排序规则；具体步骤Mapper读取结果文件，按照制表符进行切分解析出相应字段封装为SpeakBeanSpeakBean实现WritableComparable接口重写compareTo方法map()方法输出kv;key-->SpeakBean,value-->NullWritable.get()Reducer循环遍历输出Mapper代码package com.test.mr.WritableComparable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SortMapper extends Mapper<LongWritable, Text, SpeakBeanSort, NullWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] arr = value.toString().split("\t"); SpeakBeanSort bean = new SpeakBeanSort(Long.parseLong(arr[1]), Long.parseLong(arr[2]), arr[0]); context.write(bean, NullWritable.get()); } }Reducerpackage com.test.mr.WritableComparable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SortReducer extends Reducer<SpeakBeanSort, NullWritable, NullWritable, SpeakBeanSort> { @Override protected void reduce(SpeakBeanSort key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { for (NullWritable value : values) { context.write(value, key); } } }Bean对象实现WritableComparable接口package com.test.mr.WritableComparable; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class SpeakBeanSort implements WritableComparable<SpeakBeanSort> { //定义属性 private Long selfDuration;//自有内容时长 private Long thirdPartDuration;//第三方内容时长 private String deviceId;//设备id private Long sumDuration;//总时长 //准备一个空参构造 public SpeakBeanSort() { } //序列化方法:就是把内容输出到网络或者文本中 @Override public void write(DataOutput out) throws IOException { out.writeLong(selfDuration); out.writeLong(thirdPartDuration); out.writeUTF(deviceId); out.writeLong(sumDuration); } //有参构造 public SpeakBeanSort(Long selfDuration, Long thirdPartDuration, String deviceId) { this.selfDuration = selfDuration; this.thirdPartDuration = thirdPartDuration; this.deviceId = deviceId; this.sumDuration = this.selfDuration + this.thirdPartDuration; } //反序列化方法 @Override public void readFields(DataInput in) throws IOException { this.selfDuration = in.readLong();//自由时长 this.thirdPartDuration = in.readLong();//第三方时长 this.deviceId = in.readUTF();//设备id this.sumDuration = in.readLong();//总时长 } public Long getSelfDuration() { return selfDuration; } public void setSelfDuration(Long selfDuration) { this.selfDuration = selfDuration; } public Long getThirdPartDuration() { return thirdPartDuration; } public void setThirdPartDuration(Long thirdPartDuration) { this.thirdPartDuration = thirdPartDuration; } public String getDeviceId() { return deviceId; } public void setDeviceId(String deviceId) { this.deviceId = deviceId; } public Long getSumDuration() { return sumDuration; } public void setSumDuration(Long sumDuration) { this.sumDuration = sumDuration; } //为了方便观察数据，重写toString()方法 @Override public String toString() { return selfDuration + "\t" + thirdPartDuration + "\t" + deviceId + "\t" + sumDuration; } //重写compareTo方法，此处是比较一个字段如果比较2个字段就是二次排序 @Override public int compareTo(SpeakBeanSort o) { int result; // 按照总流量大小，倒序排列 if (sumDuration > o.getSumDuration()) { result = -1; }else if (sumDuration < o.getSumDuration()) { result = 1; }else { result = 0; } return result; } }Driverpackage com.test.mr.WritableComparable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class SortDriver { public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException { // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 指定本程序的jar包所在的本地路径 job.setJarByClass(SortDriver.class); // 3 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(SortMapper.class); job.setReducerClass(SortReducer.class); // 4 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(SpeakBeanSort.class); job.setMapOutputValueClass(NullWritable.class); // 5 指定最终输出的数据的kv类型 job.setOutputKeyClass(NullWritable.class); job.setOutputValueClass(SpeakBeanSort.class); // 6 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }总结自定义对象作为Map的key输出时，需要实现WritableComparable接口，排序：重写compareTo()方法，序列以及反序列化方法再次理解reduce()方法的参数；reduce()方法是map输出的kv中key相同的kv中的v组成一个集合调用一次reduce()方法，选择遍历values得到所有的key.默认reduceTask数量是1个；对于全局排序需要保证只有一个reduceTask分区排序（默认的分区规则，区内有序）GroupingComparatorGroupingComparator是mapreduce当中reduce端的一个功能组件，主要的作用是决定哪些数据作为一组，调用一次reduce的逻辑，默认是每个不同的key，作为多个不同的组，每个组调用一次reduce逻辑，我们可以自定义GroupingComparator实现不同的key作为同一个组，调用一次reduce逻辑。需求原始数据订单id商品id成交金额Order_0000001Pdt_01222.8Order_0000001Pdt_0525.8Order_0000002Pdt_03522.8Order_0000002Pdt_04122.4Order_0000002Pdt_05722.4Order_0000003Pdt_01232.8需要求出每一个订单中成交金额最大的一笔交易。实现思路Mapper1）读取一行文本数据，切分出每个字段；2）订单id和金额封装为一个Bean对象，Bean对象的排序规则指定为先按照订单Id排序，订单Id相等再按照金额降序排；3）map()方法输出kv;key-->bean对象，value-->NullWritable.get()；Shuffle1）指定分区器，保证相同订单id的数据去往同个分区（自定义分区器）2）指定GroupingComparator，分组规则指定只要订单Id相等则认为属于同一组；Reduce每个reduce()方法写出一组key的第一个参考代码OrderBeanOrderBean定义两个字段，一个字段是orderId，第二个字段是金额（注意金额一定要使用Double或者DoubleWritable类型，否则没法按照金额顺序排序）排序规则指定为先按照订单Id排序，订单Id相等再按照金额降序排！！package com.test.mr.group; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class OrderBean implements WritableComparable<OrderBean> { private String orderId; private Double price; @Override public int compareTo(OrderBean o) { //比较订单id的排序顺序 int i = this.orderId.compareTo(o.orderId); if(i == 0) { //如果订单id相同，则比较金额，金额大的排在前面 i = -this.price.compareTo(o.price); } return i; } @Override public void write(DataOutput out) throws IOException { out.writeUTF(orderId); out.writeDouble(price); } @Override public void readFields(DataInput in ) throws IOException { this.orderId = in .readUTF(); this.price = in .readDouble(); } public OrderBean() { } public OrderBean(String orderId, Double price) { this.orderId = orderId; this.price = price; } public String getOrderId() { return orderId; } public void setOrderId(String orderId) { this.orderId = orderId; } public Double getPrice() { return price; } public void setPrice(Double price) { this.price = price; } @Override public String toString() { return orderId + "\t" + price; } }自定义分区器保证ID相同的订单去往同个分区最终去往同一个Reduce中package com.test.mr.group; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Partitioner; public class CustomPartitioner extends Partitioner<OrderBean, NullWritable> { @Override public int getPartition(OrderBean orderBean, NullWritable nullWritable, int i) { //自定义分区，将相同订单id的数据发送到同一个reduce里面去 return (orderBean.getOrderId().hashCode() & Integer.MAX_VALUE) % i; } }自定义GroupingComparator保证id相同的订单进入一个分组中，进入分组的数据已经是按照金额降序排序。reduce()方法取出第一个即是金额最高的交易package com.test.mr.group; import org.apache.hadoop.io.WritableComparable; import org.apache.hadoop.io.WritableComparator; public class CustomGroupingComparator extends WritableComparator { //将我们自定义的OrderBean注册到我们自定义的CustomGroupIngCompactor当中来 //表示我们的分组器在分组的时候，对OrderBean这一种类型的数据进行分组 //传入作为key的bean的class类型，以及制定需要让框架做反射获取实例对象 public CustomGroupingComparator() { super(OrderBean.class, true); } @Override public int compare(WritableComparable a, WritableComparable b) { OrderBean first = (OrderBean) a; OrderBean second = (OrderBean) b; final int i = first.getOrderId().compareTo(second.getOrderId()); if (i == 0) { System.out.println(first.getOrderId() + "----" + second.getOrderId()); } return i; } }Mapperpackage com.test.mr.group; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class GroupMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //读取一行文本，然后切分 String[] arr = value.toString().split("t"); OrderBean bean = new OrderBean(arr[0], Double.parseDouble(arr[2])); context.write(bean, NullWritable.get()); } }Reducerpackage com.test.mr.group; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class GroupReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> { @Override protected void reduce(OrderBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { context.write(key, NullWritable.get()); } }Driverpackage com.test.mr.group; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class GroupDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1 获取配置信息，或者job对象实例 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2 指定本程序的jar包所在的本地路径 job.setJarByClass(GroupDriver.class); // 3 指定本业务job要使用的mapper/Reducer业务类 job.setMapperClass(GroupMapper.class); job.setReducerClass(GroupReducer.class); // 4 指定mapper输出数据的kv类型 job.setMapOutputKeyClass(OrderBean.class); job.setMapOutputValueClass(NullWritable.class); // 5 指定最终输出的数据的kv类型 job.setOutputKeyClass(OrderBean.class); job.setOutputValueClass(NullWritable.class); // 6 指定job的输入原始文件所在目录 FileInputFormat.setInputPaths(job, new Path("E:\teach\hadoop框架\资料\data\GroupingComparator")); FileOutputFormat.setOutputPath(job, new Path("E:\group-out")); // 7 指定分区器，指定分组比较器，设置reducetask数量 job.setPartitionerClass(CustomPartitioner.class); job.setGroupingComparatorClass(CustomGroupingComparator.class); job.setNumReduceTasks(2); // 8 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行 Boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }MapReduce读取和输出数据InputFormat运行MapReduce程序时，输入的文件格式包括:基于行的日志文件、二进制格式文件、数据库表等。那么，针对不同的数据类型，MapReduce是如何读取这些数据的呢？InputFormat是MapReduce框架用来读取数据的类。InputFormat常见子类包括:TextInputFormat （普通文本文件，MR框架默认的读取实现类型）KeyValueTextInputFormat（读取一行文本数据按照指定分隔符，把数据封装为kv类型）NLineInputF ormat(读取数据按照行数进行划分分片)CombineTextInputFormat(合并小文件，避免启动过多MapTask任务)自定义InputFormatCombineTextInputFormat案例MR框架默认的TextInputFormat切片机制按文件划分切片，文件无论多小，都是单独一个切片，然后由一个MapTask处理，如果有大量小文件，就对应的会生成并启动大量的 MapTask，而每个MapTask处理的数据量很小大量时间浪费在初始化资源启动收回等阶段，这种方式导致资源利用率不高。CombineTextInputFormat用于小文件过多的场景，它可以将多个小文件从逻辑上划分成一个切片，这样多个小文件就可以交给一个MapTask处理，提高资源利用率。需求：将输入数据中的多个小文件合并为一个切片处理使用方式：运行WordCount案例，准备多个小文件// 如果不设置InputFormat，它默认用的是TextInputFormat.class job.setInputFormatClass(CombineTextInputFormat.class); //虚拟存储切片最大值设置4m CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);验证切片数量的变化CombineTextInputFormat切片原理：切片生成过程分为两部分：虚拟存储过程和切片过程假设设置setMaxInputSplitSize值为4M四个小文件：1.txt→2M；2.txt→7M；3.txt→0.3M；4.txt→8.2M虚拟存储过程：把输入目录下所有文件大小，依次和设置的setMaxInputSplitSize值进行比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。比如如setMaxInputSplitSize值为4M，输入文件大小为8.02M，则先逻辑上分出一个4M的块。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的非常小的虚拟存储文件，所以将剩余的4.02M文件切分成（2.01M和2.01M）两个文件。1.txt→2M;2M<4M;一个块；2.txt→7M;7M>4M,但是不大于两倍，均匀分成两块；两块：每块3.5M；3.txt→0.3M;0.3<4M ,0.3M<4M ,一个块4.txt→8.2M;大于最大值且大于两倍；一个4M的块，剩余4.2M分成两块，每块2.1M所有块信息：2M，3.5M，3.5M，0.3M，4M，2.1M，2.1M 共7个虚拟存储块。切片过程判断虚拟存储的文件大小是否大于setMaxInputSplitSize值，大于等于则单独形成一个切片。如果不大于则跟下一个虚拟存储文件进行合并，共同形成一个切片。按照之前输入文件：有4个小文件大小分别为2M、7M、0.3M以及8.2M这四个小文件，则虚拟存储之后形成7个文件块，大小分别为：2M，3.5M，3.5M，0.3M，4M，2.1M，2.1M最终会形成3个切片，大小分别为：（2+3.5）M，（3.5+0.3+4）M，（2.1+2.1）M注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。自定义InputFormatHDFS还是MapReduce，在处理小文件时效率都非常低，但又难免面临处理大量小文件的场景，此时，就需要有相应解决方案。可以自定义InputFormat实现小文件的合并。需求：将多个小文件合并成一个SequenceFile文件（SequenceFile文件是Hadoop用来存储二进制形式的key-value对的文件格式），SequenceFile里面存储着多个文件，存储的形式为文件路径+名称为key，文件内容为value。结果：得到一个合并了多个小文件的SequenceFile文件整体思路：定义一个类继承FileInputFormat（TextInputFormat是它的子类）重写isSplitable()指定为不可切分；重写createRecordReader()方法，创建自己的RecorderReader（实现自定义读取数据）对象改变默认读取数据方式，实现一次读取一个完整文件作为kv输出；Driver指定使用的InputFormat类型代码参考：自定义InputFormatpackage com.test.mr.inputformat; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.JobContext; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import java.io.IOException; public class CustomFileInputformat extends FileInputFormat<Text, BytesWritable>{ //文件不可切分 @Override protected Boolean isSplitable(JobContext context, Path filename) { return false; } //获取自定义RecordReader对象用来读取数据 @Override public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { CustomRecordReader recordReader = new CustomRecordReader(); recordReader.initialize(split, context); return recordReader; } }自定义RecordReaderpackage com.test.mr.inputformat; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.input.FileSplit; import java.io.IOException; public class CustomRecordReader extends RecordReader<Text, BytesWritable> { private Configuration configuration; //切片 private FileSplit split; //是否读取到内容的标识符 private Boolean isProgress = true; //输出的kv private BytesWritable value = new BytesWritable(); private Text k = new Text(); @Override public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException { //获取到文件切片以及配置文件对象 this.split = (FileSplit) split; configuration = context.getConfiguration(); } @Override public Boolean nextKeyValue() throws IOException, InterruptedException { if (isProgress) { // 1 定义缓存区 byte[] contents = new byte[(int) split.getLength()]; FileSystem fs = null; FSDataInputStream fis = null; try { // 2 获取文件系统 Path path = split.getPath(); fs = path.getFileSystem(configuration); // 3 读取数据 fis = fs.open(path); // 4 读取文件内容 IOUtils.readFully(fis, contents, 0, contents.length); // 5 输出文件内容 value.set(contents, 0, contents.length); // 6 获取文件路径及名称 String name = split.getPath().toString(); // 7 设置输出的key值 k.set(name); } catch (Exception e) { } finally { IOUtils.closeStream(fis); } isProgress = false; return true; } return false; } @Override public Text getCurrentKey() throws IOException, InterruptedException { //返回key return k; } @Override public BytesWritable getCurrentValue() throws IOException, InterruptedException { //返回value return value; } @Override public float getProgress() throws IOException, InterruptedException { return 0; } @Override public void close() throws IOException { } }Mapperpackage com.test.mr.inputformat; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class SequenceFileMapper extends Mapper<Text, BytesWritable,Text,BytesWritable> { @Override protected void map(Text key, BytesWritable value, Context context) throws IOException, InterruptedException { //读取内容直接输出 context.write(key, value); } }Reducerpackage com.test.mr.inputformat; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> { @Override protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException { //输出value值，其中只有一个BytesWritable 所以直接next取出即可 context.write(key, values.iterator().next()); } }Driverpackage com.test.mr.inputformat; import com.test.mr.partition.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat; import java.io.IOException; public class SequenceFileDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(SequenceFileDriver.class); job.setMapperClass(SequenceFileMapper.class); job.setReducerClass(SequenceFileReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(BytesWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(BytesWritable.class); job.setInputFormatClass(CustomFileInputformat.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); Boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }验证输出结果OutputFormatOutputFormat:是MapReduce输出数据的基类，所有MapReduce的数据输出都实现了OutputFormat抽象类。下面介绍几种常见的OutputFormat子类TextOutputFormat默认的输出格式是TextOutputFormat，它把每条记录写为文本行。它的键和值可以是任意类型，因为TextOutputFormat调用toString()方 法把它们转换为字符串。SequenceFileOutputFormat将SequenceFileOutputFormat输出作为后续MapReduce任务的输入，这是一种好的输出格式，因为它的格式紧凑，很容易被压缩。自定义OutputFormat需求分析：要在一个MapReduce程序中根据数据的不同输出两类结果到不同目录，这类输出需求可以通过自定义OutputFormat来实现。实现步骤：自定义一个类继承FileOutputFormat。改写RecordWriter，改写输出数据的方法write()。需求：网络请求日志数据http://www.baidu.comhttp://www.google.comhttp://cn.bing.comhttps://www.zhaopin.comhttp://www.lagou.comhttps://www.liepin.comhttp://www.sohu.comhttp://www.sina.comhttp://www.sin2a.comhttp://www.sin2desa.comhttp://www.sindsafa.com输出结果：lagou.log文件，内容为：http://www.lagou.comother.log文件，内容为：http://www.baidu.comhttp://www.google.comhttp://cn.bing.comhttps://www.zhaopin.comhttps://www.liepin.comhttp://www.sohu.comhttp://www.sina.comhttp://www.sin2a.comhttp://www.sin2desa.comhttp://www.sindsafa.com参考代码：Mapperpackage com.test.mr.outputformat; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class OutputMapper extends Mapper<LongWritable, Text,Text, NullWritable>{ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value, NullWritable.get()); } }Reducerpackage com.test.mr.outputformat; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class OutputReducer extends Reducer<Text, NullWritable,Text,NullWritable>{ @Override protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { context.write(key,NullWritable.get() ); } }OutputFormatpackage com.test.mr.outputformat; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class CustomOutputFormat extends FileOutputFormat<Text, NullWritable> { @Override public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException { //获取文件系统对象 FileSystem fs = FileSystem.get(context.getConfiguration()); //指定输出数据的文件 Path lagouPath = new Path("e:/lagou.log"); Path otherLog = new Path("e:/other.log"); //获取输出流 FSDataOutputStream lagouOut = fs.create(lagouPath); FSDataOutputStream otherOut = fs.create(otherLog); return new CustomWriter(lagouOut, otherOut); } }RecordWriterpackage com.test.mr.outputformat; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; import java.io.IOException; public class CustomWriter extends RecordWriter<Text, NullWritable> { private FSDataOutputStream lagouOut; private FSDataOutputStream otherOut; public CustomWriter(FSDataOutputStream lagouOut,FSDataOutputStream otherOut){ this.lagouOut=lagouOut; this.otherOut=otherOut; } @Override public void write(Text key, NullWritable value) throws IOException, InterruptedException { // 判断是否包含“lagou”输出到不同文件 if (key.toString().contains("lagou")) { lagouOut.write(key.toString().getBytes()); lagouOut.write("rn".getBytes()); } else { otherOut.write(key.toString().getBytes()); otherOut.write("rn".getBytes()); } } @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { IOUtils.closeStream(lagouOut); IOUtils.closeStream(otherOut); } }Driverpackage com.test.mr.outputformat; import com.test.mr.partition.*; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class OutputDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(OutputDriver.class); job.setMapperClass(OutputMapper.class); job.setReducerClass(OutputReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(NullWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(NullWritable.class); job.setOutputFormatClass(CustomOutputFormat.class); FileInputFormat.setInputPaths(job, new Path("e:/click_log.data")); // outputformat继承自fileoutputformat 而fileoutputformat要输出一个_SUCCESS文件，所以，在这还得指定一个输出目录 FileOutputFormat.setOutputPath(job, new Path("e:/click_log/output")); Boolean flag = job.waitForCompletion(true); System.exit(flag ? 0 : 1); } }验证结果是否已把数据分别输出到不同的目录中shuffle阶段数据的压缩机制hadoop当中支持的压缩算法数据压缩有两大好处，节约磁盘空间，加速数据在网络和磁盘上的传输。我们可以使用 hadoop checknative 命令来查看我们编译之后的hadoop支持的各种压缩，如果出现openssl为false，那么就在线安装一下依赖包。安装opensslyum install -y openssl-devel压缩格式hadoop自带算法文件扩展名是否可切分换成压缩格式后，原来的程序是否需要修改DEFLATE是，直接使用DEFLATE.deflate否和文本处理一样，不需要修改Gzip是，直接使用DEFLATE.gz否和文本处理一样，不需要修改bzip2是，直接使用bzip2.bz2是和文本处理一样，不需要修改LZO否，需要安装LZO.lzo是需要建索引，还需要指定输入格式Snappy否，需要安装Snappy.snappy否和文本处理一样，不需要修改为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器压缩格式对应的编码/解码器DEFLATEorg.apache.hadoop.io.compress.DefaultCodecgziporg.apache.hadoop.io.compress.GzipCodecbzip2org.apache.hadoop.io.compress.BZip2CodecLZOcom.hadoop.compression.lzo.LzopCodecSnappyorg.apache.hadoop.io.compress.SnappyCodec常见压缩方式对比分析压缩算法原始文件大小压缩后的文件大小压缩速度解压缩速度gzip8.3GB1.8GB17.5MB/s58MB/sbzip28.3GB1.1GB2.4MB/s9.5MB/sLZO-bset8.3GB2GB4MB/s60.6MB/sLZO8.3GB2.9GB49.3MB/S74.6MB/s压缩位置Map输入端压缩此处使用压缩文件作为Map的输入数据，无需显示指定编解码方式，Hadoop会自动检查文件扩展名，如果压缩方式能够匹配，Hadoop就会选择合适的编解码方式对文件进行压缩和解压。Map输出端压缩Shuffle是Hadoop MR过程中资源消耗最多的阶段，如果有数据量过大造成网络传输速度缓慢，可以考虑使用压缩Reduce端输出压缩输出的结果数据使用压缩能够减少存储的数据量，降低所需磁盘的空间，并且作为第二个MR的输入时可以复用压缩。压缩配置方式在驱动代码中通过Configuration直接设置使用的压缩方式，可以开启Map输出和Reduce输出压缩// 设置map阶段压缩 Configuration configuration = new Configuration(); configuration.set("mapreduce.map.output.compress","true"); configuration.set("mapreduce.map.output.compress.codec","org.apache.hadoop.io.compress.SnappyCodec"); // 设置reduce阶段的压缩 configuration.set("mapreduce.output.fileoutputformat.compress","true"); configuration.set("mapreduce.output.fileoutputformat.compress.type","RECORD"); configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");配置mapred-site.xml(修改后分发到集群其它节点，重启Hadoop集群)，此种方式对运行在集群的所有MR任务都会执行压缩。<property> <name>mapreduce.output.fileoutputformat.compress</name> <value>true</value> </property> <property> <name>mapreduce.output.fileoutputformat.compress.type</name> <value>RECORD</value> </property> <property> <name>mapreduce.output.fileoutputformat.compress.codec</name> <value>org.apache.hadoop.io.compress.SnappyCodec</value> </property>压缩案例需求：使用snappy压缩方式压缩WordCount案例的输出结果数据具体实现：在驱动代码中添加压缩配置configuration.set("mapreduce.output.fileoutputformat.compress","true"); configuration.set("mapreduce.output.fileoutputformat.compress.type","RECORD"); configuration.set("mapreduce.output.fileoutputformat.compress.codec","org.apache.hadoop.io.compress.SnappyCodec");重新打成jar包，提交集群运行，验证输出结果是否已进行了snappy压缩
HDFS简介HDFS（全称：Hadoop Distribute File System，Hadoop 分布式文件系统）是 Hadoop 核心组成，是分布式存储服务。分布式文件系统横跨多台计算机，在大数据时代有着广泛的应用前景，它们为存储和处理超大规模数据提供所需的扩展能力。HDFS是分布式文件系统中的一种。HDFS的重要概念HDFS 通过统一的命名空间目录树来定位文件； 另外，它是分布式的，由很多服务器联合起来实现其功能，集群中的服务器有各自的角色（分布式本质是拆分，各司其职）；典型的 Master/Slave 架构HDFS 的架构是典型的 Master/Slave 结构。HDFS集群往往是一个NameNode（HA架构会有两个NameNode,联邦机制）+多个DataNode组成；NameNode是集群的主节点，DataNode是集群的从节点。分块存储（block机制）HDFS 中的文件在物理上是分块存储（block）的，块的大小可以通过配置参数来规定；Hadoop2.x版本中默认的block大小是128M；命名空间（NameSpace）HDFS 支持传统的层次型文件组织结构。用户或者应用程序可以创建目录，然后将文件保存在这些目录里。文件系统名字空间的层次结构和大多数现有的文件系统类似：用户可以创建、删除、移动或重命名文件。Namenode 负责维护文件系统的名字空间，任何对文件系统名字空间或属性的修改都将被 Namenode 记录下来。HDFS提供给客户单一个抽象目录树，访问形式：hdfs://namenode的hostname:port/test/inputhdfs://linux121:9000/test/inputNameNode元数据管理我们把目录结构及文件分块位置信息叫做元数据。NameNode的元数据记录每一个文件所对应的block信息（block的id,以及所在的DataNode节点的信息）DataNode数据存储文件的各个 block 的具体存储管理由 DataNode 节点承担。一个block会有多个DataNode来存储，DataNode会定时向NameNode来汇报自己持有的block信息。副本机制为了容错，文件的所有 block 都会有副本。每个文件的 block 大小和副本系数都是可配置的。应用程序可以指定某个文件的副本数目。副本系数可以在文件创建的时候指定，也可以在之后改变。副本数量默认是3个。一次写入，多次读出HDFS 是设计成适应一次写入，多次读出的场景，且不支持文件的随机修改。 （支持追加写入，不只支持随机更新）正因为如此，HDFS 适合用来做大数据分析的底层存储服务，并不适合用来做网盘等应用（修改不方便，延迟大，网络开销大，成本太高）HDFS架构NameNode(nn):Hdfs集群的管理者，Master维护管理Hdfs的名称空间（NameSpace）维护副本策略记录文件块（Block）的映射信息负责处理客户端读写请求DataNode:NameNode下达命令，DataNode执行实际操作，Slave节点。保存实际的数据块负责数据块的读写Client:客户端上传文件到HDFS的时候，Client负责将文件切分成Block,然后进行上传请求NameNode交互，获取文件的位置信息读取或写入文件，与DataNode交互Client可以使用一些命令来管理HDFS或者访问HDFSHDFS客户端操作Shell 命令行操作HDFS基本语法bin/hadoop fs 具体命令 OR bin/hdfs dfs 具体命令命令大全hdfs dfsHDFS命令演示1）启动Hadoop集群（方便后续的测试）start-dfs.sh start-yarn.sh2）-help：输出这个命令参数hadoop fs -help rm3）-ls: 显示目录信息hadoop fs -ls /4）-mkdir：在HDFS上创建目录hadoop fs -mkdir -p /test/bigdata5）-moveFromLocal：从本地剪切粘贴到HDFStouch hadoop.txt hadoop fs -moveFromLocal ./hadoop.txt /test/bigdata6）-appendToFile：追加一个文件到已经存在的文件末尾touch hdfs.txt vi hdfs.txt输入namenode datanode block replicationhadoop fs -appendToFile hdfs.txt /test/bigdata/hadoop.txt7）-cat：显示文件内容hadoop fs -cat /test/bigdata/hadoop.txt8）-chgrp 、-chmod、-chown：Linux文件系统中的用法一样，修改文件所属权限hadoop fs -chmod 666 /test/bigdata/hadoop.txt hadoop fs -chown root:root /test/bigdata/hadoop.txt9）-copyFromLocal：从本地文件系统中拷贝文件到HDFS路径去hadoop fs -copyFromLocal README.txt /10）-copyToLocal：从HDFS拷贝到本地hadoop fs -copyToLocal /test/bigdata/hadoop.txt ./11）-cp ：从HDFS的一个路径拷贝到HDFS的另一个路径hadoop fs -cp /test/bigdata/hadoop.txt /hdfs.txt12）-mv：在HDFS目录中移动文件hadoop fs -mv /hdfs.txt /test/bigdata/13）-get：等同于copyToLocal，就是从HDFS下载文件到本地hadoop fs -get /test/bigdata/hadoop.txt ./14）-put：等同于copyFromLocalhadoop fs -mkdir -p /user/root/test/ vim yarn.txt输入resourcemanager nodemanagerhadoop fs -put ./yarn.txt /user/root/test/15）-tail：显示一个文件的末尾hadoop fs -tail /user/root/test/yarn.txt16）-rm：删除文件或文件夹hadoop fs -rm /user/root/test/yarn.txt17）-rmdir：删除空目录hadoop fs -mkdir /test hadoop fs -rmdir /test18）-du统计文件夹的大小信息hadoop fs -du -s -h /user/root/test hadoop fs -du -h /user/root/test19）-setrep：设置HDFS中文件的副本数量hadoop fs -setrep 10 /test/bigdata/hadoop.txt这里设置的副本数只是记录在NameNode的元数据中，是否真的会有这么多副本，还得看DataNode的数量。因为目前只有3台设备，最多也就3个副本，只有节点数的增加到10台时，副本数才能达到10。JAVA客户端客户端环境准备将Hadoop-2.9.2安装包解压到非中文路径（例如：E:\hadoop-2.9.2）。配置HADOOP_HOME环境变量配置Path环境变量。创建一个Maven工程ClientDemo导入相应的依赖坐标+日志配置文件<dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>RELEASE</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>2.9.2</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.9.2</version> </dependency> </dependencies>为了便于控制程序运行打印的日志数量，需要在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，文件内容：log4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n创建包名：com.test.hdfs创建HdfsClient类public class HdfsClient{ @Test public void testMkdirs() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); // 配置在集群上运行 // configuration.set("fs.defaultFS", "hdfs://linux121:9000"); // FileSystem fs = FileSystem.get(configuration); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 创建目录 fs.mkdirs(new Path("/test")); // 3 关闭资源 fs.close(); } }遇到问题：如果不指定操作HDFS集群的用户信息，默认是获取当前操作系统的用户信息，出现权限被拒绝的问题，报错如下：HDFS的API操作上传文件编写源代码@Test public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); configuration.set("dfs.replication", "2"); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 上传文件 fs.copyFromLocalFile(new Path("e:/test.txt"), new Path("/test.txt")); // 3 关闭资源 fs.close(); System.out.println("end"); }将hdfs-site.xml拷贝到项目的根目录下<?xml version="1.0" encoding="UTF-8"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> <configuration> <property> <name>dfs.replication</name> <value>1</value> </property> </configuration>参数优先级参数优先级排序：代码中设置的值>用户自定义配置文件>服务器的默认配置下载文件@Test public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 执行下载操作 // boolean delSrc 指是否将原文件删除 // Path src 指要下载的文件路径 // Path dst 指将文件下载到的路径 // boolean useRawLocalFileSystem 是否开启文件校验 fs.copyToLocalFile(false, new Path("/test.txt"), new Path("e:/test_copy.txt"), true); // 3 关闭资源 fs.close(); }删除文件/文件夹@Test public void testDelete() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 执行删除 fs.delete(new Path("/api_test/"), true); // 3 关闭资源 fs.close(); }查看文件名称、权限、长度、块信息@Test public void testListFiles() throws IOException, InterruptedException, URISyntaxException{ // 1获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 获取文件详情 RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true); while(listFiles.hasNext()){ LocatedFileStatus status = listFiles.next(); // 输出详情 // 文件名称 System.out.println(status.getPath().getName()); // 长度 System.out.println(status.getLen()); // 权限 System.out.println(status.getPermission()); // 分组 System.out.println(status.getGroup()); // 获取存储的块信息 BlockLocation[] blockLocations = status.getBlockLocations(); for (BlockLocation blockLocation : blockLocations) { // 获取块存储的主机节点 String[] hosts = blockLocation.getHosts(); for (String host : hosts) { System.out.println(host); } } System.out.println("-----------华丽的分割线----------"); } // 3 关闭资源 fs.close(); }文件夹判断@Test public void testListStatus() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件配置信息 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 判断是文件还是文件夹 FileStatus[] listStatus = fs.listStatus(new Path("/")); for (FileStatus fileStatus : listStatus) { // 如果是文件 if (fileStatus.isFile()) { System.out.println("f:"+fileStatus.getPath().getName()); }else { System.out.println("d:"+fileStatus.getPath().getName()); } } // 3 关闭资源 fs.close(); }I/O流操作HDFS以上我们使用的API操作都是HDFS系统框架封装好的。我们自己也可以采用IO流的方式实现文件的上传和下载。文件上传需求：把本地e盘上的test.txt文件上传到HDFS根目录编写代码@Test public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException { // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 创建输入流 FileInputStream fis = new FileInputStream(new File("e:/test.txt")); // 3 获取输出流 FSDataOutputStream fos = fs.create(new Path("/test_io.txt")); // 4 流对拷 IOUtils.copyBytes(fis, fos, configuration); // 5 关闭资源 IOUtils.closeStream(fos); IOUtils.closeStream(fis); fs.close(); }文件下载需求：从HDFS上下载test.txt文件到本地e盘上编写代码@Test public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 获取输入流 FSDataInputStream fis = fs.open(new Path("/test_io.txt")); // 3 获取输出流 FileOutputStream fos = new FileOutputStream(new File("e:/test_io_copy.txt")); // 4 流的对拷 IOUtils.copyBytes(fis, fos, configuration); // 5 关闭资源 IOUtils.closeStream(fos); IOUtils.closeStream(fis); fs.close(); }seek 定位读取需求：将HDFS上的test.txt的内容在控制台输出两次编写代码@Test public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{ // 1 获取文件系统 Configuration configuration = new Configuration(); FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root"); // 2 打开输入流,读取数据输出到控制台 FSDataInputStream in = null; try{ in= fs.open(new Path("/test.txt")); IOUtils.copyBytes(in, System.out, 4096, false); //从头再次读取 in.seek(0); IOUtils.copyBytes(in, System.out, 4096, false); }finally { IOUtils.closeStream(in); } }注意:windows解压安装Hadoop后，在调用相关API操作HDFS集群时可能会报错，这是由于Hadoop安装缺少windows操作系统相关文件所致，如下图：解决方案：百度找到winutils.exe拷贝放到windows系统Hadoop安装目录的bin目录下即可！！HDFS文件系统权限问题：hdfs的文件权限机制与linux系统的文件权限机制类似！！r:read w:write x:execute 权限x对于文件表示忽略，对于文件夹表示是否有权限访问其内容如果linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS当中的owner就是zhangsanHDFS文件权限的目的，防止好人做错事，而不是阻止坏人做坏事。HDFS相信你告诉我你是谁，你就是谁！！解决方案：1）指定用户信息获取FileSystem对象2）关闭HDFS集群权限校验vim hdfs-site.xml添加如下属性<property> <name>dfs.permissions</name> <value>true</value> </property>修改完成之后要分发到其它节点，同时要重启HDFS集群3）基于HDFS权限本身比较鸡肋的特点，我们可以彻底放弃HDFS的权限校验，如果生产环境中我们可以考虑借助kerberos以及sentry等安全框架来管理大数据集群安全。所以我们直接修改HDFS的根目录权限为777hadoop fs -chmod -R 777 /HDFS读写解析HDFS读数据流程客户端通过Distributed FileSystem向NameNode请求下载文件，NameNode通过查询元数据，找到文件块所在的DataNode地址。挑选一台DataNode（就近原则，然后随机）服务器，请求读取数据。DataNode开始传输数据给客户端（从磁盘里面读取数据输入流，以Packet为单位来做校验）。客户端以Packet为单位接收，先在本地缓存，然后写入目标文件。HDFS写数据流程客户端通过Distributed FileSystem模块向NameNode请求上传文件，NameNode检查目标文件是否已存在，父目录是否存在。NameNode返回是否可以上传。客户端请求第一个 Block上传到哪几个DataNode服务器上。NameNode返回3个DataNode节点，分别为dn1、dn2、dn3。客户端通过FSDataOutputStream模块请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。dn1、dn2、dn3逐级应答客户端。客户端开始往dn1上传第一个Block（先从磁盘读取数据放到一个本地内存缓存），以Packet为单位，dn1收到一个Packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个确认队列等待确认。当一个Block传输完成之后，客户端再次请求NameNode上传第二个Block的服务器。（重复执行3-7步）。验证Packet代码@Test public void testUploadPacket() throws IOException { //1 准备读取本地文件的输入流 FileInputStream in = new FileInputStream(new File("e:/test.txt")); //2 准备好写出数据到hdfs的输出流 FSDataOutputStream out = fs.create(new Path("/test.txt"), new Progressable() { //这个progress方法就是每传输64KB（packet）就会执行一次 public void progress() { System.out.println("&"); } }); //3 实现流拷贝，默认关闭流选项是true，所以会自动关闭 IOUtils.copyBytes(in, out, configuration); //4 关流 可以再次关闭也可以不关了 }NN与2NNHDFS元数据管理机制问题1：NameNode如何管理和存储元数据？计算机中存储数据两种：内存或者是磁盘元数据存储磁盘：存储磁盘无法面对客户端对元数据信息的任意的快速低延迟的响应，但是安全性高元数据存储内存：元数据存放内存，可以高效的查询以及快速响应客户端的查询请求，数据保存在内存，如果断电，内存中的数据全部丢失。解决方案：内存+磁盘；NameNode内存+FsImage的文件（磁盘）问题2：磁盘和内存中元数据如何划分？两个数据一模一样，还是两个数据合并到一起才是一份完整的数据呢？一模一样：client如果对元数据进行增删改操作，需要保证两个数据的一致性。FsImage文件操作起来效率也不高。两个合并=完整数据：NameNode引入了一个edits文件（日志文件：只能追加写入）edits文件记录的是client的增删改操作，不再选择让NameNode把数据dump出来形成FsImage文件（这种操作是比较消耗资源）。元数据管理流程图第一阶段：NameNode启动第一次启动NameNode格式化后，创建Fsimage和Edits文件。如果不是第一次启动，直接加载编辑日志和镜像文件到内存。客户端对元数据进行增删改的请求。NameNode记录操作日志，更新滚动日志。NameNode在内存中对数据进行增删改。第二阶段：Secondary NameNode工作Secondary NameNode询问NameNode是否需要CheckPoint。直接带回NameNode是否执行检查点操作结果。Secondary NameNode请求执行CheckPoint。NameNode滚动正在写的Edits日志。将滚动前的编辑日志和镜像文件拷贝到Secondary NameNode。Secondary NameNode加载编辑日志和镜像文件到内存，并合并。生成新的镜像文件fsimage.chkpoint。拷贝fsimage.chkpoint到NameNode。NameNode将fsimage.chkpoint重新命名成fsimage。Fsimage与Edits文件解析NameNode在执行格式化之后，会在/opt/test/servers/hadoop-2.9.2/data/tmp/dfs/name/current目录下产生如下文件Fsimage文件：是namenode中关于元数据的镜像，一般称为检查点，这里包含了HDFS文件系统所有目录以及文件相关信息（Block数量，副本数量，权限等信息）Edits文件 ：存储了客户端对HDFS文件系统所有的更新操作记录，Client对HDFS文件系统所有的更新操作都会被记录到Edits文件中（不包括查询操作）seen_txid：该文件是保存了一个数字，数字对应着最后一个Edits文件名的数字VERSION：该文件记录namenode的一些版本号信息，比如：CusterId，namespaceID等Fsimage文件内容官方地址https://hadoop.apache.org/docs/r2.9.2/hadoop-project-dist/hadoophdfs/HdfsImageViewer.html查看oiv和oev命令hdfsoiv Offline Image Viewer View a Hadoop fsimage INPUTFILE using the specified PROCESSOR,saving the results in OUTPUTFILE.oev Offline edits viewer Parse a Hadoop edits log file INPUT_FILE and save results in OUTPUT_FILE基本语法hdfs oiv -p 文件类型(xml) -i 镜像文件 -o 转换后文件输出路径案例实操cd /opt/test/servers/hadoop-2.9.2/data/tmp/dfs/name/current hdfs oiv -p XML -i fsimage_0000000000000000265 -o /opt/test/servers/fsimage.xml cat /opt/test/servers/fsimage.xml<?xml version="1.0"?> <fsimage> <version> <layoutVersion>-63</layoutVersion> <onDiskVersion>1</onDiskVersion> <oivRevision>826afbeae31ca687bc2f8471dc841b66ed2c6704</oivRevision> </version> <NameSection> <namespaceId>1393381414</namespaceId> <genstampV1>1000</genstampV1> <genstampV2>1024</genstampV2> <genstampV1Limit>0</genstampV1Limit> <lastAllocatedBlockId>1073741848</lastAllocatedBlockId> <txid>265</txid> </NameSection> <INodeSection> <inode> <id>16398</id> <type>DIRECTORY</type> <name>history</name> <mtime>1592376391028</mtime> <permission>root:supergroup:0777</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16399</id> <type>DIRECTORY</type> <name>done_intermediate</name> <mtime>1592375256896</mtime> <permission>root:supergroup:1777</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16400</id> <type>DIRECTORY</type> <name>root</name> <mtime>1592378079208</mtime> <permission>root:supergroup:0777</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16413</id> <type>FILE</type> <name>job_1592375222804_0001-1592375231176-root-word+count- 1592375281926-1-1-SUCCEEDED-default-1592375261492.jhist</name> <replication>3</replication> <mtime>1592375282039</mtime> <atime>1592375281980</atime> <preferredBlockSize>134217728</preferredBlockSize> <permission>root:supergroup:0777</permission> <blocks> <block> <id>1073741834</id> <genstamp>1010</genstamp> <numBytes>33584</numBytes> </block> </blocks> <storagePolicyId>0</storagePolicyId> </inode> <inode> <id>16414</id> <type>FILE</type> <name>job_1592375222804_0001_conf.xml</name> <replication>3</replication> <mtime>1592375282121</mtime> <atime>1592375282053</atime> <preferredBlockSize>134217728</preferredBlockSize> <permission>root:supergroup:0777</permission> <blocks> <block> <id>1073741835</id> <genstamp>1011</genstamp> <numBytes>196027</numBytes> </block> </blocks> <storagePolicyId>0</storagePolicyId> </inode> <inode> <id>16415</id> <type>DIRECTORY</type> <name>done</name> <mtime>1592376776670</mtime> <permission>root:supergroup:0777</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16427</id> <type>DIRECTORY</type> <name>logs</name> <mtime>1592378009623</mtime> <permission>root:root:0770</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16428</id> <type>DIRECTORY</type> <name>application_1592376944601_0001</name> <mtime>1592378045481</mtime> <permission>root:root:0770</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16430</id> <type>DIRECTORY</type> <name>wcoutput</name> <mtime>1592378037463</mtime> <permission>root:supergroup:0755</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16436</id> <type>FILE</type> <name>part-r-00000</name> <replication>3</replication> <mtime>1592378037264</mtime> <atime>1592378037074</atime> <preferredBlockSize>134217728</preferredBlockSize> <permission>root:supergroup:0644</permission> <blocks> <block> <id>1073741842</id> <genstamp>1018</genstamp> <numBytes>43</numBytes> </block> </blocks> <storagePolicyId>0</storagePolicyId> </inode> <inode> <id>16445</id> <type>FILE</type> <name>linux123_39919</name> <replication>3</replication> <mtime>1592378045469</mtime> <atime>1592378045331</atime> <preferredBlockSize>134217728</preferredBlockSize> <permission>root:root:0640</permission> <blocks> <block> <id>1073741848</id> <genstamp>1024</genstamp> <numBytes>56910</numBytes> </block> </blocks> <storagePolicyId>0</storagePolicyId> </inode> <inode> <id>16446</id> <type>DIRECTORY</type> <name>0617</name> <mtime>1592387393490</mtime> <permission>root:supergroup:0755</permission> <nsquota>-1</nsquota> <dsquota>-1</dsquota> </inode> <inode> <id>16449</id> <type>FILE</type> <name>banzhang.txt</name> <replication>1</replication> <mtime>1592388309046</mtime> <atime>1592388309026</atime> <preferredBlockSize>134217728</preferredBlockSize> <permission>root:supergroup:0644</permission> <storagePolicyId>0</storagePolicyId> </inode> </INodeSection> </fsimage>问题：Fsimage中为什么没有记录块所对应DataNode？在内存元数据中是有记录块所对应的dn信息，但是fsimage中就剔除了这个信息；HDFS集群在启动的时候会加载image以及edits文件，block对应的dn信息都没有记录，集群启动时会有一个安全模式（safemode）,安全模式就是为了让dn汇报自己当前所持有的block信息给nn来补全元数据。后续每隔一段时间dn都要汇报自己持有的block信息。Edits文件内容基本语法hdfs oev -p 文件类型 -i编辑日志 -o 转换后文件输出路径案例实操hdfs oev -p XML -i edits_0000000000000000266-0000000000000000267 -o /opt/test/servers/hadoop-2.9.2/edits.xml cat /opt/test/servers/hadoop-2.9.2/edits.xml<?xml version="1.0" encoding="UTF-8"?> <EDITS> <EDITS_VERSION>-63</EDITS_VERSION> <RECORD> <OPCODE>OP_START_LOG_SEGMENT</OPCODE> <DATA> <TXID>113</TXID> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>114</TXID> <SRC>/wcoutput/_SUCCESS</SRC> <MODE>493</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>115</TXID> <SRC>/wcoutput/part-r-00000</SRC> <MODE>493</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>116</TXID> <SRC>/wcoutput</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>117</TXID> <SRC>/wcoutput/_SUCCESS</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>118</TXID> <SRC>/wcoutput/part-r-00000</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_DELETE</OPCODE> <DATA> <TXID>119</TXID> <LENGTH>0</LENGTH> <PATH>/wcoutput/part-r-00000</PATH> <TIMESTAMP>1592377324171</TIMESTAMP> <RPC_CLIENTID></RPC_CLIENTID> <RPC_CALLID>-2</RPC_CALLID> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>120</TXID> <SRC>/</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>121</TXID> <SRC>/tmp</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>122</TXID> <SRC>/tmp/hadoop-yarn</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>123</TXID> <SRC>/tmp/hadoop-yarn/staging</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>124</TXID> <SRC>/tmp/hadoop-yarn/staging/history</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>125</TXID> <SRC>/tmp/hadoop-yarn/staging/history/done</SRC> <MODE>511</MODE> </RECORD> <RECORD> <OPCODE>OP_SET_PERMISSIONS</OPCODE> <DATA> <TXID>126</TXID> <SRC>/tmp/hadoop-yarn/staging/history/done/2020</SRC> <MODE>511</MODE> </RECORD> </EDITS>备注：Edits中只记录了更新相关的操作，查询或者下载文件并不会记录在内！！问题：NameNode启动时如何确定加载哪些Edits文件呢？nn启动时需要加载fsimage文件以及那些没有被2nn进行合并的edits文件，nn如何判断哪些edits已经被合并了？可以通过fsimage文件自身的编号来确定哪些已经被合并。checkpoint周期hdfs-default.xml <property> <name>dfs.namenode.checkpoint.period</name> <value>3600</value> </property> <property> <name>dfs.namenode.checkpoint.txns</name> <value>1000000</value> <description>操作动作次数</description> </property> <property> <name>dfs.namenode.checkpoint.check.period</name> <value>60</value> <description> 1分钟检查一次操作次数</description> </property>NN故障处理NameNode故障后，HDFS集群就无法正常工作，因为HDFS文件系统的元数据需要由NameNode来管理维护并与Client交互，如果元数据出现损坏和丢失同样会导致NameNode无法正常工作进而HDFS文件系统无法正常对外提供服务。如果元数据出现丢失损坏如何恢复呢？将2NN的元数据拷贝到NN的节点下此种方式会存在元数据的丢失。搭建HDFS的HA（高可用）集群，解决NN的单点故障问题！！（借助Zookeeper实现HA，一个Active的NameNode,一个是Standby的NameNode）Hadoop的限额与归档以及集群安全模式HDFS文件限额配置HDFS文件的限额配置允许我们以文件大小或者文件个数来限制我们在某个目录下上传的文件数量或者文件内容总量，以便达到我们类似百度网盘网盘等限制每个用户允许上传的最大的文件的量数量限额#创建hdfs文件夹 hdfs dfs -mkdir -p /user/root/test # 给该文件夹下面设置最多上传两个文件，上传文件，发现只能上传一个文件 hdfs dfsadmin -setQuota 2 /user/root/test # 清除文件数量限制 hdfs dfsadmin -clrQuota /user/root/test 空间大小限额# 限制空间大小4KB hdfs dfsadmin -setSpaceQuota 4k /user/root/test #上传超过4Kb的文件大小上去提示文件超过限额 hdfs dfs -put /export/softwares/xxx.tar.gz /user/root/test #清除空间限额 hdfs dfsadmin -clrSpaceQuota /user/root/test #查看hdfs文件限额数量 hdfs dfs -count -q -h /user/root/testHDFS的安全模式安全模式是HDFS所处的一种特殊状态，在这种状态下，文件系统只接受读数据请求，而不接受删除、修改等变更请求。在NameNode主节点启动时，HDFS首先进入安全模式，DataNode在启动的时候会向NameNode汇报可用的block等状态，当整个系统达到安全标准时，HDFS自动离开安全模式。如果HDFS出于安全模式下，则文件block不能进行任何的副本复制操作，因此达到最小的副本数量要求是基于DataNode启动时的状态来判定的，启动时不会再做任何复制（从而达到最小副本数量要求），HDFS集群刚启动的时候，默认30S钟的时间是出于安全期的，只有过了30S之后，集群脱离了安全期，然后才可以对集群进行操作。hdfs dfsadmin -safemodeHadoop归档技术主要解决HDFS集群存在大量小文件的问题！！由于大量小文件会占用NameNode的内存，因此对于HDFS来说存储大量小文件造成NameNode内存资源的浪费！Hadoop存档文件HAR文件，是一个更高效的文件存档工具，HAR文件是由一组文件通过archive工具创建而来，在减少了NameNode的内存使用的同时，可以对文件进行透明的访问，通俗来说就是HAR文件对NameNode来说是一个文件减少了内存的浪费，对于实际操作处理文件依然是一个一个独立的文件。案例启动YARN集群start-yarn.sh归档文件把/user/test/input目录里面的所有文件归档成一个叫input.har的归档文件，并把归档后文件存储到/user/test/output路径下。hadoop archive -archiveName input.har –p /user/root/input /user/root/output查看归档hadoop fs -lsr /user/root/output/input.har hadoop fs -lsr har:///user/root/output/input.har解归档文件hadoop fs -cp har:///user/root/output/input.har/* /user/root
软件和操作系统版本Hadoop框架是采用Java语言编写，需要java环境（jvm）JDK版本：JDK8版本集群：知识点学习：统一使用vmware虚拟机虚拟三台linux节点，linux操作系统：Centos7项目阶段：统一使用云服务器，5台云服务器节点Hadoop搭建方式单机模式：单节点模式，非集群，生产不会使用这种方式单机伪分布式模式：单节点，多线程模拟集群的效果，生产不会使用这种方式完全分布式模式：多台节点，真正的分布式Hadoop集群的搭建（生产环境建议使用这种方式）虚拟机环境准备三台虚拟机（静态IP，关闭防火墙，修改主机名，配置免密登录，集群时间同步）在/opt目录下创建文件夹创建软件安装包存放目录和软件安装目录mkdir -p /opt/test/software mkdir -p /opt/test/serversHadoop下载地址：https://archive.apache.org/dist/hadoop/common/hadoop-2.9.2/Hadoop官网地址：http://hadoop.apache.org/上传hadoop压缩包到/opt/test/software集群规划框架linux121linux122linux123HDFSNameNode、DataNodeDataNodeSecondaryNameNode、DataNodeYARNNodeManagerNodeManagerNodeManager、ResourceManager安装Hadoop登录linux121节点；进入/opt/test/software，解压安装文件到/opt/test/serverstar -zxvf hadoop-2.9.2.tar.gz -C /opt/test/servers查看是否解压成功ll /opt/test/servers/hadoop-2.9.2添加Hadoop到环境变量vim /etc/profile##HADOOP_HOME export HADOOP_HOME=/opt/test/servers/hadoop-2.9.2 export PATH=$PATH:$HADOOP_HOME/bin export PATH=$PATH:$HADOOP_HOME/sbin使环境变量生效source /etc/profile验证hadoophadoop version校验结果：hadoop目录drwxr-xr-x. 2 root root 194 Nov 13 2018 bindrwxr-xr-x. 3 root root 20 Nov 13 2018 etcdrwxr-xr-x. 2 root root 106 Nov 13 2018 includedrwxr-xr-x. 3 root root 20 Nov 13 2018 libdrwxr-xr-x. 2 root root 239 Nov 13 2018 libexec-rw-r--r--. 1 root root 106210 Nov 13 2018 LICENSE.txt-rw-r--r--. 1 root root 15917 Nov 13 2018 NOTICE.txt-rw-r--r--. 1 root root 1366 Nov 13 2018 README.txtdrwxr-xr-x. 3 root root 4096 Nov 13 2018 sbindrwxr-xr-x. 4 root root 31 Nov 13 2018 share1）bin目录:对Hadoop进行操作的相关命令，如hadoop，hdfs等2）etc目录：Hadoop的配置文件目录，如hdfs-site.xml，core-site.xml等3）lib目录：Hadoop本地库（解压缩的依赖）4）sbin目录：存放的是Hadoop集群启动停止相关脚本，命令5）share目录：Hadoop的一些jar，官方案例jar，文档等集群配置Hadoop集群配置 = HDFS集群配置 + MapReduce集群配置 + Yarn集群配置HDFS集群配置1）将JDK路径明确配置给HDFS（修改hadoop-env.sh）2）指定NameNode节点以及数据存储目录（修改core-site.xml）3）指定SecondaryNameNode节点（修改hdfs-site.xml）4）指定DataNode从节点（修改etc/hadoop/slaves文件，每个节点配置信息占一行）MapReduce集群配置1）将JDK路径明确配置给MapReduce（修改mapred-env.sh）2）指定MapReduce计算框架运行Yarn资源调度框架（修改mapred-site.xml）Yarn集群配置1）将JDK路径明确配置给Yarn（修改yarn-env.sh）2）指定ResourceManager老大节点所在计算机节点（修改yarn-site.xml）3）指定NodeManager节点（会通过slaves文件内容确定）集群配置具体步骤：HDFS集群配置cd /opt/test/servers/hadoop-2.9.2/etc/hadoop配置：hadoop-env.sh将JDK路径明确配置给HDFSvim hadoop-env.shexport JAVA_HOME=/opt/test/servers/jdk1.8.0_231指定NameNode节点以及数据存储目录（修改core-site.xml）vim core-site.xml <property> <name>fs.defaultFS</name> <value>hdfs://linux121:9000</value> </property> <property> <name>hadoop.tmp.dir</name> <value>/opt/test/servers/hadoop-2.9.2/data/tmp</value> </property>core-site.xml的默认配置：https://hadoop.apache.org/docs/r2.9.2/hadoop-project-dist/hadoop-common/coredefault.xml指定secondarynamenode节点(修改hdfs-site.xml)vim hdfs-site.xml <property> <name>dfs.namenode.secondary.http-address</name> <value>linux123:50090</value> </property> <property> <name>dfs.replication</name> <value>3</value> </property>官方默认配置https://hadoop.apache.org/docs/r2.9.2/hadoop-project-dist/hadoop-hdfs/hdfsdefault.xml指定datanode从节点(修改slaves文件，每个节点配置信息占一行)vim slaveslinux121 linux122 linux123注意：该文件中添加的内容结尾不允许有空格，文件中不允许有空行。MapReduce集群配置指定MapReduce使用的jdk路径（修改mapred-env.sh）vim mapred-env.shexport JAVA_HOME=/opt/test/servers/jdk1.8.0_231指定MapReduce计算框架运行Yarn资源调度框架(修改mapred-site.xml)mv mapred-site.xml.template mapred-site.xml vim mapred-site.xml <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property>mapred-site.xml默认配置https://hadoop.apache.org/docs/r2.9.2/hadoop-mapreduce-client/hadoop-mapreduceclient-core/mapred-default.xmlYarn集群配置指定jdk路径（修改yarn-env.sh）vim yarn-env.shexport JAVA_HOME=/opt/test/servers/jdk1.8.0_231指定ResourceMnager的master节点信息(修改yarn-site.xml)vim yarn-site.xml <property> <name>yarn.resourcemanager.hostname</name> <value>linux123</value> </property> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property>yarn-site.xml的默认配置https://hadoop.apache.org/docs/r2.9.2/hadoop-yarn/hadoop-yarn-common/yarndefault.xml指定NodeManager节点（slaves文件已修改）注意：Hadoop安装目录所属用户和所属用户组信息，默认是501 dialout，而我们操作Hadoop集群的用户使用的是虚拟机的root用户，所以为了避免出现信息混乱，修改Hadoop安装目录所属用户和用户组！！chown -R root:root /opt/test/servers/hadoop-2.9.2分发配置编写集群分发脚本rsync-scriptrsync 远程同步工具rsync主要用于备份和镜像。具有速度快、避免复制相同内容和支持符号链接的优点。rsync和scp区别：用rsync做文件的复制要比scp的速度快，rsync只对差异文件做更新。scp是把所有文件都复制过去。1）基本语法rsync -rvl $pdir/$fname $user@$host:$pdir/$fname命令 选项参数 要拷贝的文件路径/名称 目的用户@主机:目的路径/名称2）选项参数说明选项功能-r递归-v显示复制过程-l拷贝符号连接rsync案例1）三台虚拟机安装rsync (执行安装需要保证机器联网)yum install -y rsync2）把linux121机器上的/opt/test/software目录同步到linux122服务器的root用户下的/opt/目录rsync -rvl /opt/test/software/ root@linux122:/opt/test/software集群分发脚本编写1）需求：循环复制文件到集群所有节点的相同目录下rsync命令原始拷贝：rsync -rvl /opt/module root@linux123:/opt/2）期望脚本脚本+要同步的文件名称3）说明：在/usr/local/bin这个目录下存放的脚本，root用户可以在系统任何地方直接执行。4）脚本实现在/usr/local/bin目录下创建文件rsync-script，文件内容如下：touch rsync-script vim rsync-script在文件中编写shell代码#!/bin/bash #1 获取命令输入参数的个数，如果个数为0，直接退出命令 paramnum=$# if((paramnum==0)); then echo no params; exit; fi #2 根据传入参数获取文件名称 p1=$1 file_name=`basename $p1` echo fname=$file_name #3 获取输入参数的绝对路径 pdir=`cd -P $(dirname $p1); pwd` echo pdir=$pdir #4 获取用户名称 user=`whoami` #5 循环执行rsync for((host=121; host<124; host++)); do echo ------------------- linux$host -------------- rsync -rvl $pdir/$file_name $user@linux$host:$pdir done修改脚本 rsync-script 具有执行权限chmod 777 rsync-script调用脚本形式：rsync-script 文件名称rsync-script /home/root/bin调用脚本分发Hadoop安装目录到其它节点rsync-script /opt/test/servers/hadoop-2.9.2shell脚本的编写即使不掌握，但是看到别人编写的脚本要能读懂！！启动集群注意：如果集群是第一次启动，需要在Namenode所在节点格式化NameNode，非第一次不用执行格式化Namenode操作！！单节点启动hadoop namenode -format格式化后创建的文件：/opt/test/servers/hadoop-2.9.2/data/tmp/dfs/name/current在linux121上启动NameNodehadoop-daemon.sh start namenode jps在linux121、linux122以及linux123上分别启动DataNodehadoop-daemon.sh start datanode jpsweb端查看Hdfs界面http://linux121:50070/dfshealth.html#tab-overview查看HDFS集群正常节点：Yarn集群单节点启动在linux123上yarn-daemon.sh start resourcemanager jps在linux121和linux122上yarn-daemon.sh start nodemanager jps思考：Hadoop集群每次需要一个一个节点的启动，如果节点数增加到成千上万个怎么办？集群群起如果已经单节点方式启动了Hadoop,可以先停止之前的启动的Namenode与Datanode进程,如果之前Namenode没有执行格式化，这里需要执行格式化!!!!hadoop namenode -format启动HDFS在linux121上sbin/start-dfs.sh jps可以在122和123上使用jps命令查看HDFS是否启动启动YARN在linux123上sbin/start-yarn.sh注意：NameNode和ResourceManger不是在同一台机器，不能在NameNode上启动 YARN，应该在ResouceManager所在的机器上启动YARN。Hadoop集群启动停止命令汇总各个服务组件逐一启动/停止（1）分别启动/停止HDFS组件hadoop-daemon.sh start / stop namenode / datanode / secondarynamenode（2）启动/停止YARNyarn-daemon.sh start / stop resourcemanager / nodemanager各个模块分开启动/停止（配置ssh是前提）常用（1）整体启动/停止HDFSstart-dfs.sh / stop-dfs.sh（2）整体启动/停止YARNstart-yarn.sh / stop-yarn.sh集群测试HDFS 分布式存储初体验从linux本地文件系统上传下载文件验证HDFS集群工作正常本地root目录创建一个文件hdfs dfs -mkdir -p /test/input cd /root vim test.txthello hdfs上传linxu文件到hdfshdfs dfs -put /root/test.txt /test/input从hdfs下载文件到linux本地hdfs dfs -get /test/input/test.txtMapReduce 分布式计算初体验在HDFS文件系统根目录下面创建一个wcinput文件夹hdfs dfs -mkdir /wcinput在/root/目录下创建一个wc.txt文件(本地文件系统)cd /root/ touch wc.txt vi wc.txt在文件中输入如下内容hadoop mapreduce yarn hdfs hadoop mapreduce mapreduce yarn test test test保存退出:wq!上传wc.txt到Hdfs目录/wcinput下hdfs dfs -put wc.txt /wcinput回到Hadoop目录/opt/test/servers/hadoop-2.9.2执行程序hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.9.2.jar wordcount /wcinput /wcoutput注意：这里的wcinput必须存在，wcoutput必须不存在，要让程序自己创建查看结果hdfs dfs -cat /wcoutput/part-r-00000hadoop 2hdfs 1test 3mapreduce 3yarn 2配置历史服务器在Yarn中运行的任务产生的日志数据不能查看，为了查看程序的历史运行情况，需要配置一下历史日志服务器。具体配置步骤如下：配置mapred-site.xmlvi mapred-site.xml在该文件里面增加如下配置。 <property> <name>mapreduce.jobhistory.address</name> <value>linux121:10020</value> </property> <property> <name>mapreduce.jobhistory.webapp.address</name> <value>linux121:19888</value> </property>分发mapred-site.xml到其它节点rsync-script mapred-site.xml启动历史服务器因为前面第一步中在121上配置的历史服务器，所以必须在121上启动sbin/mr-jobhistory-daemon.sh start historyserver查看历史服务器是否启动jps查看JobHistoryhttp://linux121:19888/jobhistory配置日志的聚集日志聚集：应用(Job)运行完成以后，将应用运行日志信息从各个task汇总上传到HDFS系统上。日志聚集功能好处：可以方便的查看到程序运行详情，方便开发调试。注意：开启日志聚集功能，需要重新启动NodeManager 、ResourceManager和HistoryManager。开启日志聚集功能具体步骤如下：配置yarn-site.xmlvi yarn-site.xml在该文件里面增加如下配置。 <property> <name>yarn.log-aggregation-enable</name> <value>true</value> </property> <property> <name>yarn.log-aggregation.retain-seconds</name> <value>604800</value> </property>分发yarn-site.xml到集群其它节点rsync-script yarn-site.xml121上关闭NodeManager 、ResourceManager和HistoryManagersbin/yarn-daemon.sh stop resourcemanager sbin/yarn-daemon.sh stop nodemanager sbin/mr-jobhistory-daemon.sh stop historyserver121上启动NodeManager 、ResourceManager和HistoryManagersbin/yarn-daemon.sh start resourcemanager sbin/yarn-daemon.sh start nodemanager sbin/mr-jobhistory-daemon.sh start historyserver121上删除HDFS上已经存在的输出文件bin/hdfs dfs -rm -R /wcoutput121上执行WordCount程序hadoop jar share/hadoop/mapreduce/hadoopmapreduce-examples-2.9.2.jar wordcount /wcinput /wcoutput查看日志，如图所示http://linux121:19888/jobhistory
Hadoop=HDFS(分布式文件系统)+MapReduce(分布式计算框架)+Yarn(资源协调框架)+Common模块Hadoop HDFS：（Hadoop Distribute File System ）一个高可靠、高吞吐量的分布式文件系统比如：100T数据存储该怎么设计？Google发表的论文翻译成中文即“分而治之”分：拆分→数据切割，100T数据拆分为10G一个数据块由一个电脑节点存储这个数据块。数据切割、制作副本、分散储存图中涉及到几个角色NameNode（nn）：存储文件的元数据，比如文件名、文件目录结构、文件属性（生成时间、副本数、文件权限），以及每个文件的块列表和块所在的DataNode等。SecondaryNameNode（2nn）：辅助NameNode更好的工作，用来监控HDFS状态的辅助后台程序，每隔一段时间获取HDFS元数据快照。DataNode（dn）：在本地文件系统存储文件块数据，以及块数据的校验。注意：NN，2NN,DN这些既是角色名称，进程名称，代指电脑节点名称！！Hadoop MapReduce：一个分布式的离线并行计算框架拆解任务、分散处理、汇整结果MapReduce计算 = Map阶段 + Reduce阶段Map阶段就是“分”的阶段，并行处理输入数据；Reduce阶段就是“合”的阶段，对Map阶段结果进行汇总；Hadoop YARN：作业调度与集群资源管理的框架计算资源协调Yarn中有如下几个主要角色，同样，既是角色名、也是进程名，也指代所在计算机节点名称。ResourceManager(rm)：处理客户端请求、启动/监控ApplicationMaster、监控NodeManager、资源分配与调度；NodeManager(nm)：单个节点上的资源管理、处理来自ResourceManager的命令、处理来自ApplicationMaster的命令；ApplicationMaster(am)：数据切分、为应用程序申请资源，并分配给内部任务、任务监控与容错。Container：对任务运行环境的抽象，封装了CPU、内存等多维资源以及环境变量、启动命令等任务运行相关的信息。ResourceManager是老大，NodeManager是小弟，ApplicationMaster是计算任务专员。Hadoop Common：支持其他模块的工具模块（Configuration、RPC、序列化机制、日志操作）