分布式NoSQL列存储数据库Hbase_高级思想（八）
文章目录
[*]分布式NoSQL列存储数据库Hbase_高级思想（八）
[*]
[*]知识点01：课程回顾
[*]知识点02：课程目标
[*]知识点03：Hbase读写流程：写入流程
[*]知识点04：Hbase读写流程：meta表
[*]知识点05：Hbase读写流程：整体写入流程
[*]知识点06：Hbase读写流程：整体读取流程
[*]知识点07：LSM模型：设计思想
[*]知识点08：LSM模型：Flush
[*]知识点09：LSM模型：Compaction
[*]知识点10：Region分裂Split设计及规则
[*]知识点11：MapReduce回顾
[*]知识点12：MR集成Hbase：读Hbase规则
[*]知识点13：MR集成Hbase：读Hbase实现
[*]知识点14：MR集成Hbase：写Hbase规则
[*]知识点15：MR集成Hbase：写Hbase实现
[*]附录一：Maven依赖
分布式NoSQL列存储数据库Hbase_高级思想（八）
知识点01：课程回顾
[*] 为什么要构建二级索引？
[*]Hbase使用Rowkey作为唯一索引，只有使用Rowkey前缀进行查询，才是索引查询
[*]导致大部分的查询都是不走索引，性能比较差
[*] 为什么二级索引能提高查询性能？
[*]使用走两次索引代替全表扫描
[*]先走索引查询索引表，获取原表的rowkey
[*]再根据原表的rowkey查询原表的数据
[*] Phoenix为什么可以实现二级索引？
[*] Phoenix底层封装了大量的协处理器
create index indexName on Tbname(col)
[*] 什么是全局索引？
[*] 创建全局索引，会自动构建一张索引表
[*] 索引表结构
[*]Rowkey：索引字段+原表的rowkey
[*]列：占位置x
[*] 特点
[*]如果查询字段或者查询条件不是索引字段，就不会走索引
[*] 应用
[*] 适合于读多写少
[*] 所有索引的构建都会阻塞原表的请求
[*] 什么是覆盖索引？
[*] 创建覆盖索引，会自动构建一张索引表
[*] 索引表结构
[*]Rowkey：索引字段+原表的rowkey
[*]列：将include中的列放入索引表
[*] 特点
[*] 如果查询字段或者查询条件不是索引字段，就不会走索引
[*] 如果查询的字段在索引表中，直接从索引表返回结果
[*] 什么是本地索引？
[*] 创建覆盖索引，会自动基于原表构建一个列族来实现索引存储
[*] 原表的数据中：多了一个索引列族
[*] 特点
[*]不论查询字段是否是索引字段，都会走索引
[*]将索引与数据存储在同一台RegionServer，提高索引读写性能
[*] 注意
[*] 本地索引会修改原数据表，对于本地索引只能使用Phoenix来操作表的数据
[*] 盐表不能使用本地索引
[*] HMaster的功能是什么？
[*]管理节点
[*]NameNode
[*]管理从节点：DataNode
[*]管理元数据
[*]NameNode接受DN数据块的汇报：保证数据安全
[*]管理元数据：将管理类的元数据存储在ZK中
[*]管理从节点：HRegionServer
[*]管理Region：分配、校验
[*] HRegionServer的功能是什么？
[*]负责存储Hbase中Region的数据
[*]提供分布式内存
[*]WAL：预写日志
[*]Region：分区
[*]Store
[*]MemStore：写缓存
[*]BlockCache：读缓存
[*]StoreFIle
[*] HDFS与Zookeeper的功能是什么？
[*]HDFS：分布式磁盘
[*]存储StoreFIle文件对应的HFILE
[*]Zookeeper
[*]辅助选举
[*]存储管理元数据
知识点02：课程目标
[*]Hbase完整的读写流程
[*]存储结构
[*]HRegionServer：存储节点
[*]Region：用于实现分布式表
[*]Store：用于划分列的存储
[*]MemStore：写内存
[*]StoreFIle：用于存储大量数据的持久化文件
[*]写数据过程
[*]读数据过程
[*]表的元数据检索【重点】
[*]读写数据时，如何获取表对应的元数据的？
[*]Hbase设计模型：LSM模型【了解】
[*]先写内存：Log
[*]将内存写入磁盘：Flush
[*]对磁盘数据进行合并：Compact
[*]Hbase特有的一个功能：Split
[*]MapReduce如何读写Hbase【了解】
[*]使用输入类和输出类：特点以及要求==【重点关注：Spark读写Hbase】==
[*]TableInputFormat
[*]TableOutputFormat
知识点03：Hbase读写流程：写入流程
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CP4uDLNM-1616659900232)(20210325_分布式NoSQL列存储数据库Hbase（八）.assets/image-20210323223403690.png)]
[*] 目标
[*] 掌握Hbase数据写入的流程
[*] 当执行一条Put操作，数据是如何写入Hbase的？
put 表名rowkey列族：列值
[*] 分析
[*]step1：根据表名获取这张表对应的所有Region的信息
[*] step2：根据Rowkey判断具体写入哪个Region
[*]step3：将put操作提交给这个Region所在的RegionServer
[*] step4：RegionServer将数据写入Region，根据列族判断写入哪个Store
[*]step5：将数据写入MemStore中
[*] 总结
[*]表名：决定了这条数据要写入哪些region中
[*]Rowkey：决定了这条数据具体写入哪个Region中
[*]列族：决定了写入这个region哪个Store中
知识点04：Hbase读写流程：meta表
[*] 目标
[*]了解hbase:meta表的存储内容及功能
[*]问题1：如何知道这张表对应的region有哪些？
[*]问题2：如何知道每个Region的范围的？
[*]问题3：如何知道Region所在的RegionServer地址的？
[*] 分析
[*]肯定有个地方存储了表与Region的关系以及Region的信息
[*]存储每张表对应的所有region关系
[*]每个region的范围和RegionServer地址
[*] 实现
[*] Hbase自带的两张系统表
[*]hbase:namespace：存储了Hbase中所有namespace的信息
[*]hbase:meta：存储了表的元数据
[*] hbase:meta表结构
[*] Rowkey：每张表每个Region的名称
itcast:t4,eeeeeeee,1616123941870.ba3bf6d78ce9432ea4cd42a3829142b2.
表名,startKey，时间戳，Region的唯一id
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mx2aqWAs-1616659900234)(20210325_分布式NoSQL列存储数据库Hbase（八）.assets/image-20210325091441577.png)]
[*]Hbase中每张表的每个region对应元数据表中的一个Rowkey
[*] 列
[*] info:regioninfo
#Region的名称
NAME => 'itcast:t4,eeeeeeee,1616123941870.ba3bf6d78ce9432ea4cd42a3829142b2.',
#Region的范围
STARTKEY => 'eeeeeeee', ENDKEY => ''
[*] info:server
#Region所在的RegionServer的地址
value=node3:16020
[*] 总结
[*]功能：存储了整个Hbase中每张表每个Region的信息
[*]名称
[*]起始范围
[*]RegionServer地址
[*]step1：只要知道表名，就能获取这表表对应的所有Region信息
[*]step2：根据Region的范围与Rowkey做比较，就能知道要具体写入哪个region
[*]step3：请求这个region对应的regionServer地址，写入Region即可
知识点05：Hbase读写流程：整体写入流程
[*] 目标
[*]掌握Hbase写入数据的整体流程
[*] 分析
putns:tbname,rowkey,cf:col,value
[*]step1：获取元数据
[*]表的元数据：hbase:meta表中
[*]问题：如果要往一张表写入数据，必须先读meta表，如何知道meta的地址？
[*]解决：请求zk，meta表的地址存储在zk中
[*]step2：找到对应的Region
[*]step3：写入数据
[*] 实现
[*] step1：获取元数据
[*] 客户端请求Zookeeper，获取meta表所在的regionserver的地址
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g6i0EPc8-1616659900235)(20210325_分布式NoSQL列存储数据库Hbase（八）.assets/image-20210325092843114.png)]
[*] 读取meta表的数据：获取所有表的元数据
[*] step2：找到对应的Region
[*]根据meta表中的元数据，找到表对应的所有的region
[*]根据region的范围和写入的Rowkey，判断需要写入具体哪一个Region
[*]根据region的Regionserver的地址，请求对应的RegionServer
[*] step3：写入数据
[*]请求RegionServer写入对应Region：根据Region的名称来指定写入哪个Region
[*]根据列族判断写入哪一个具体的Store
[*]先写入WAL：Hlog预写日志中
[*]写入对应Store的MemStore中
[*]MemStore
[*] 总结
[*]step1：先连接ZK，获取meta表的地址
[*]step2：读取meta表的数据，获取表的元数据
[*]step3：根据表的元数据找到对应Region的RegionServer
[*]step4:
[*]RegionServer
[*]Region
[*]WAL
[*]Store
[*]MemStore
[*]StoreFIle
知识点06：Hbase读写流程：整体读取流程
[*] 目标
[*]掌握Hbase数据读取整体流程
[*] 分析
#根据rowkey来判断读取哪个region
getns:tbname , rowkey
#读取所有Region
scan ns:tbname
[*]step1：获取元数据
[*]step2：找到对应的Region
[*]step3：读取数据
[*] 实现
[*]step1：获取元数据
[*]客户端请求Zookeeper，获取meta表所在的regionserver的地址
[*]读取meta表的数据
[*]注意：客户端会缓存meta表的数据，只有第一次会连接ZK，读取meta表的数据，缓存会定期失效，要重新缓存
[*]避免每次请求都要先连接zk，再读取meta表
[*]step2：找到对应的Region
[*]根据meta表中的元数据，找到表对应的region
[*]根据region的范围和写入的Rowkey，判断需要写入具体哪一个Region
[*]根据region的Regionserver的地址，请求对应的RegionServer
[*]step3：读取数据
[*]先查询memstore
[*]如果没有，就读取StoreFile
[*]如果查询的列族开启了缓存机制
[*]第一次
[*]先查询MEMStore
[*]没有就查询StoreFile
[*]查询到以后，将查询结果放入读缓存BlockCache
[*]第二次
[*]先查询MEMStore
[*]再查询BlockCache
[*]最后查询StoreFile
[*] 总结
[*]RegionServer
[*]Region
[*]Store
[*]Memstore
[*]BlockCache
[*]StoreFIle
知识点07：LSM模型：设计思想
[*] 目标
[*]了解LSM树结构模型(Log-Structured Merge-Tree)设计思想
[*] 分析
[*]step1：数据写入的时候，只写入内存
[*]step2：将数据在内存构建有序，当数据量大的时候，将有序的数据写入磁盘，变成一个有序的数据文件
[*]step3：基于所有有序的小文件进行合并，合并为一个整体有序的大文件
[*] 实现
[*] step1：数据写入的时候，只写入内存：MemStore
[*]通过WAL来记录内存的操作，保证数据的安全性
[*]但是数据量大的时候，内存肯定放不下，需要将数据写入磁盘
[*] step2：将数据在内存构建有序，当数据量大的时候，将有序的数据写入磁盘，变成一个有序的数据文件：Flush
[*]Hbase中基于Rowkey构建有序，Region按照范围有序划分
[*] step3：基于所有有序的小文件进行合并，合并为一个整体有序的大文件：Compaction
[*]Hbase基于Rowkey将多个StoreFile文件合并为一个StoreFile文件，加快检索的速度
[*] 符合LSM模型的数据存储系统中，一般为了性能，都只有插入，没有更新和删除
[*] 当用户做更新和删除操作时，底层并没有真正的对数据作删除和更新，而是对这个被删除和更新的数据做了标记
[*] 更新的本质：插入了一条新的数据，查询的时候，有这些标记的数据不会被显示
[*] 所有真正的更新和删除都是在Merge过程中，将标记的数据删除掉
[*] 总结
[*]step1：将数据放入内存，构建有序
[*]step2：内存达到一定阈值，将数据写入磁盘，构建有序文件
[*]step3：将多个有序文件合并为一个有序的大文件
知识点08：LSM模型：Flush
[*] 目标
[*]了解Hbase的LSM模型中Flush的设计
[*] 分析
[*]什么是Flush？
[*]将memstore中的数据写入HDFS，变成StoreFile
[*] 实现
[*] 关闭集群：自动Flush
[*] 参数配置：自动触发机制
#2.x版本之前的机制
#region的memstore的触发
#判断如果某个region中的某个memstore达到这个阈值，那么触发flush，flush这个region的所有memstore
hbase.hregion.memstore.flush.size=128M
#region的触发级别：如果没有memstore达到128，但是所有memstore的大小加在一起大于等于128*4
#触发整个region的flush
hbase.hregion.memstore.block.multiplier=4
#regionserver的触发级别：所有region所占用的memstore达到阈值，就会触发整个regionserver中memstore的溢写
#从memstore占用最多的Regin开始flush
hbase.regionserver.global.memstore.size=0.4 --RegionServer中Memstore的总大小
hbase.regionserver.global.memstore.size.lower.limit = 0.4*0.95 =0.38
#2.x版本以后的机制
#设置了一个flush的最小阈值
#memstore的判断发生了改变：max("hbase.hregion.memstore.flush.size / column_family_number",hbase.hregion.percolumnfamilyflush.size.lower.bound.min)
#如果memstore高于上面这个结果，就会被flush，如果低于这个值，就不flush，如果整个region所有的memstore都低于，全部flush
#水位线 = max（128 / 列族个数,16）,列族一般给3个 ~ 42M
#如果memstore的空间大于42,就flush，如果小于就不flush，如果都小于，全部flush
举例：3个列族，3个memstore,90/30/30 90会被Flush
举例：3个列族，3个memstore,30/30/30全部flush
hbase.hregion.percolumnfamilyflush.size.lower.bound.min=16M
#2.x中多了一种机制：In-Memory-compact,如果开启了【不为none】，会在内存中对需要flush的数据进行合并
#合并后再进行flush，将多个小文件在内存中合并后再flush
hbase.hregion.compacting.memstore.type=None|basic|eager|adaptive
[*] 总结
[*] Hbase利用Flush实现将内存数据溢写到HDFS，保持内存中不断存储最新的数据
[*] 注意：工作中一般进行手动Flush
[*] 原因：避免大量的Memstore将大量的数据同时Flush到HDFS上，占用大量的内存和磁盘的IO带宽，会影响业务
[*] 解决：手动触发，定期执行
hbase> flush 'TABLENAME'
hbase> flush 'REGIONNAME'
hbase> flush 'ENCODED_REGIONNAME'
hbase> flush 'REGION_SERVER_NAME'
[*] 封装一个文件，通过hbase shell filepath来定期的运行这个脚本
知识点09：LSM模型：Compaction
[*] 目标
[*]了解Hbase中Compaction的的设计及实现规则
[*] 分析
[*]什么是Compaction？
[*]将多个单独有序StoreFile文件进行合并，合并为整体有序的大文件，加快读取速度
[*]file1:1 2 3 4 5
[*]file2 ： 6 7 9
[*]file3 :1 8 10
[*]|| 每个文件都读取，可能读取无效的数据
[*]file：1 1 2 3 4 5 6 7 8 9 10
[*] 实现
[*] 版本功能
[*]2.0版本之前，只有StoreFile文件的合并
[*]磁盘中合并：minor compaction、major compaction
[*]2.0版本开始，内存中的数据也可以先合并后Flush
[*]内存中合并：In-memory compaction
[*]磁盘中合并：minor compaction、major compaction
[*] **In-memory compaction：**2.0版本开始新增加的功能
[*] 原理：将当前写入的数据划分segment【数据段】
[*]当数据不断写入MemStore，划分不同的segment，最终变成storefile文件
[*] 如果开启了内存合并，先将第一个segment放入一个队列中，与其他的segment进行合并
[*]合并以后的结果再进行flush
[*] 内存中合并的方式
hbase.hregion.compacting.memstore.type=None|basic|eager|adaptive
none：不开启，不合并
[*] basic（基础型）
Basic compaction策略不清理多余的数据版本，无需对cell的内存进行考核
basic适用于所有大量写模式
[*] eager（饥渴型）
eager compaction会过滤重复的数据，清理多余的版本，这会带来额外的开销
eager模式主要针对数据大量过期淘汰的场景，例如：购物车、消息队列等
[*] adaptive（适应型）
adaptive compaction根据数据的重复情况来决定是否使用eager策略
该策略会找出cell个数最多的一个，然后计算一个比例，如果比例超出阈值，则使用eager策略，否则使用basic策略
[*] minor compaction：轻量级
[*] 功能：将最早生成的几个小的StoreFile文件进行合并，成为一个大文件，不定期触发
[*] 特点
[*]只实现将多个小的StoreFile合并成一个相对较大的StoreFile，占用的资源不多
[*]不会将标记为更新或者删除的数据进行处理
[*] 属性
hbase.hstore.compaction.min=3
[*] major compaction：重量级合并
[*] 功能：将整个Store中所有StoreFile进行合并为一个StoreFile文件，整体有序的一个大文件
[*] 特点
[*]将所有文件进行合并，构建整体有序
[*]合并过程中会进行清理过期和标记为删除的数据
[*]资源消耗比较大
[*] 参数配置
hbase.hregion.majorcompaction=7天
[*] 总结
[*] Hbase通过Compaction实现将零散的有序数据合并为整体有序大文件，提高对HDFS数据的查询性能
[*] 在工作中要避免自动触发majorcompaction，影响业务
hbase.hregion.majorcompaction=0
[*] 在不影响业务的时候，手动处理，每天在业务不繁忙的时候，调度工具实现手动进行major compact
Run major compaction on passed table or pass a region row
to major compact an individual region. To compact a single
column family within a region specify the region name
followed by the column family name.
Examples:
Compact all regions in a table:
hbase> major_compact 't1'
hbase> major_compact 'ns1:t1'
Compact an entire region:
hbase> major_compact 'r1'
Compact a single column family within a region:
hbase> major_compact 'r1', 'c1'
Compact a single column family within a table:
hbase> major_compact 't1', 'c1'
Compact table with type "MOB"
hbase> major_compact 't1', nil, 'MOB'
Compact a column family using "MOB" type within a table
hbase> major_compact 't1', 'c1', 'MOB'
知识点10：Region分裂Split设计及规则
[*] 目标
[*]了解Hbase中Region的分裂功能及其策略
[*] 分析
[*]什么是Split分裂机制？
[*]当一个Region存储的数据过多，导致这个Region的负载比较高，Hbase中设定了一个Region最多存储的数据量的阈值，一旦达到阈值，允许Region分裂为两个region，老的region会下线，新的两个region对外提供服务
[*]RegionServer负责判断是否满足条件，并且负责切分Region
[*]Master负责分配两个新的region，两个新的region分配成功以后，老的region下线
[*] 实现
[*] 参数配置
#region阈值
hbase.hregion.max.filesize=10GB
#0.94之前：判断region中是否有一个storefile文件是否达到阈值，如果达到，就分裂
hbase.regionserver.region.split.policy=org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy
#0.94开始
如果region个数在0 ` 100之间
#规则：Math.min(getDesiredMaxFileSize(),initialSize * tableRegionsCount * tableRegionsCount * tableRegionsCount)
#initialSize = 128 X 2
#min(10GB,256 x region个数的3次方)
hbase.regionserver.region.split.policy=org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy
#2.x开始
#规则：return tableRegionsCount == 1? this.initialSize : getDesiredMaxFileSize();
#判断region个数是否为1，如果为1，就按照256分，如果不为1，就按照10GB来分
hbase.regionserver.region.split.policy=org.apache.hadoop.hbase.regionserver.SteppingSplitPolicy
[*] 总结
[*] Hbase通过Split策略来保证一个Region存储的数据量不会过大，通过分裂实现分摊负载，避免热点，降低故障率
[*] 注意：工作作中避免自动触发，影响集群读写，建议关闭
hbase.regionserver.region.split.policy=org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy
[*] 手动操作
split 'tableName'
split 'namespace:tableName'
split 'regionName' # format: 'tableName,startKey,id'
split 'tableName', 'splitKey'
split 'regionName', 'splitKey'
知识点11：MapReduce回顾
[*] 目标
[*]掌握MapReduce五大阶段的功能及执行过程
[*]应用：读写Hbase，通过分布式计算程序来实现分布式读写
[*]MapReduce
[*]Spark
[*]Flink
[*] 分析
[*]step1：读取数据
[*]step2：处理数据
[*]step3：输出结果
[*] 实现
[*] Input：负责整个程序的输入，由InputFormat类决定
[*] TextInputFormat：读取文件数据
1 2 3 4 5 6 7 8 9
BLK1:1 2 3
BLK2:4 5 6
BLK3:7 8 9
[*] 功能一：将读取数据划分分片
[*] TextInputFormatHDFS文件，将每一个块变成一个分片Split
BLK1:1 2 3=> split1
BLK2:4 5 6=> split2
BLK3:7 8 9=> split3
[*] 功能二：将分片中的每一条数据变成一个KV
[*]TextInputFormat：每一条数据变成一个KV对
[*]K：行的偏移量：LongWritable
[*]V：行的数据内容：Text
[*] Map：分布式计算处理，处理逻辑由map方法决定
[*]根据Input中的分片的个数，每个Split会启动一个MapTask进程，处理对应分片的数据
[*]MapTask1：split1:123
[*]MapTask2：split2:456
[*]MapTask3：split3:789
[*]处理逻辑：由自定义map方法决定处理
[*]输出新的KV
[*] Shuffle：实现全局排序、分组
[*]排序
[*]分组
[*] Reduce：分布式聚合处理，处理逻辑由reduce方法决定
[*]默认启动1个ReduceTask来实现对上一步所有分组后的数据进行聚合处理
[*]聚合逻辑：由reduce方法
[*] Output：负责整个程序的输出，由OutputFormat类决定
[*]将上一步输出的KV保存到外部系统中
[*]TextOutputFormat：将上一步的结果写入文件
[*] SQL
[*] 不做分组
select * from table
[*]Input
[*]Map
[*]Output
[*] 做分组
select 聚合函数 from table group by col
[*] Input：from
[*] Map：行和列的过滤
select id ,name from table where id = 2
[*] Shuffle：group by order by
[*] Reduce：聚合函数
[*] Output
[*]insert
[*] 代码开发规则
[*] Map类：用于在Map阶段，每个Maptask构建一个Map类实例，用于调用map方式对数据进行处理
[*]继承Mapper类，map方法
[*] Reduce类：用于在Reduce阶段，每个ReduceTask构建Reduce实例，用于调用reduce方法对数据进行处理
[*]继承Reduce类，reduce方法
[*] Driver类
[*]main：程序入口，调用run方法
[*]run：构建、配置、提交MapReduceJob
job.setMapperClass(Mapper类)
job.setReduceClass(Reducer类)
[*] 总结
[*]五大阶段的功能
[*]Input：负责输入
[*]Map：负责分布式计算
[*]Shuffle：负责分组和排序
[*]Reduce：负责聚合
[*]Output：负责输出
知识点12：MR集成Hbase：读Hbase规则
[*] 目标
[*]掌握MapReduce中读取Hbase的开发规则
[*] 分析
[*]读取由InputFormat决定
[*]TableInputFormat：负责实现读取Hbase的数据，将每个Rowkey的数据转换为一个KV对象
[*]K：ImmutableBytesWritable：Rowkey的字节对象
[*]V：Result：Rowkey的数据
[*] 实现
[*] step1：调用工具类方法，初始化Input和Map
[*]MapReduce中封装了工具类，实现读取Hbase数据
TableMapReduceUtil.initTableMapperJob
public static void initTableMapperJob(
String table,
Scan scan,
Class<? extends TableMapper> mapper,
Class<?> outputKeyClass,
Class<?> outputValueClass,
Job job
);
[*] step2：构建Map类继承TableMapper类
/**
* Extends the base <code>Mapper</code> class to add the required input key
* and value classes.
*
* @param <KEYOUT>The type of the key.
* @param <VALUEOUT>The type of the value.
* @see org.apache.hadoop.mapreduce.Mapper
*/
@InterfaceAudience.Public
public abstract class TableMapper<KEYOUT, VALUEOUT>
extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {
}
[*] 总结
[*]MapReduce读取Hbase数据的API已经封装好了，只需要调用工具类实现即可
知识点13：MR集成Hbase：读Hbase实现
[*] 目标
[*]实现从Hbase读取数据，将数据写入文件中
[*] 分析
[*]step1：使用TableInputFormat读取Hbase数据
[*]step2：使用TextOutputFormat写入文件
[*] 实现
[*] 总结
知识点14：MR集成Hbase：写Hbase规则
[*] 目标
[*]掌握MapReduce写入Hbase的开发规则
[*] 分析
[*] 输出由OutputFormat决定
[*] TableOutputFormat：负责实现将上一步的KV数据写入Hbase表中
/**
* Convert Map/Reduce output and write it to an HBase table. The KEY is ignored
* while the output value <u>must</u> be either a {@link Put} or a
* {@link Delete} instance.
*/
@InterfaceAudience.Public
public class TableOutputFormat<KEY> extends OutputFormat<KEY, Mutation>
[*]要求输出的Value类型必须为Mutation类型：Put / Delete
[*] 实现
[*] step1：调用工具类初始化Reduce和Output
[*]MapReduce中封装了工具类，实现读取Hbase数据
TableMapReduceUtil.initTableReducerJob
/**
* Use this before submitting a TableReduce job. It will
* appropriately set up the JobConf.
*
* @param tableThe output table.
* @param reducerThe reducer class to use.
* @param jobThe current job to adjust.
* @throws IOException When determining the region count fails.
*/
public static void initTableReducerJob(
String table,
Class<? extends TableReducer> reducer,
Job job
);
[*] step2：构建Reduce类继承TableReducer
/**
* Extends the basic <code>Reducer</code> class to add the required key and
* value input/output classes.
*
* @param <KEYIN>The type of the input key.
* @param <VALUEIN>The type of the input value.
* @param <KEYOUT>The type of the output key.
* @see org.apache.hadoop.mapreduce.Reducer
*/
@InterfaceAudience.Public
public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>
extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation> {
}
[*] 总结
[*]MapReduce写入Hbase数据的API已经封装好了，只需要调用工具类实现即可
知识点15：MR集成Hbase：写Hbase实现
[*] 目标
[*]实现从文件读取数据，将数据写入Hbase中
[*] 分析
[*]step1：使用TextInputFormat读取文件中的数据
[*]step2：构建Put对象，封装Rowkey以及列
[*]step3：使用TableOutputFormat将数据写入Hbase表中
[*] 实现
[*] Hbase中建表
create 'itcast:mrwrite','info'
[*] 读取文件数据
[*] 构建Put对象
[*] 初始化Reduce和Output：调用TableOutputFormat
[*] 总结
附录一：Maven依赖
<repositories>
<repository>
<id>aliyun</id>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
</repository>
</repositories>
<properties>
<hadoop.version>2.7.3</hadoop.version>
<hbase.version>2.1.2</hbase.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>${hbase.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-mapreduce</artifactId>
<version>${hbase.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-auth</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
<version>${hadoop.version}</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
</dependencies>
.version}
org.apache.hadoop
hadoop-common
h a d o o p . v e r s i o n < / v e r s i o n > < / d e p e n d e n c y > < d e p e n d e n c y > < g r o u p I d > o r g . a p a c h e . h a d o o p < / g r o u p I d > < a r t i f a c t I d > h a d o o p − m a p r e d u c e − c l i e n t − c o r e < / a r t i f a c t I d > < v e r s i o n > {hadoop.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version> hadoop.version</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop−mapreduce−client−core</artifactId><version>{hadoop.version}
org.apache.hadoop
hadoop-auth
h a d o o p . v e r s i o n < / v e r s i o n > < / d e p e n d e n c y > < d e p e n d e n c y > < g r o u p I d > o r g . a p a c h e . h a d o o p < / g r o u p I d > < a r t i f a c t I d > h a d o o p − h d f s < / a r t i f a c t I d > < v e r s i o n > {hadoop.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version> hadoop.version</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop−hdfs</artifactId><version>{hadoop.version}
commons-io
commons-io
2.6
页:
[1]