【4.分布式计算】hadoop-hbase/druid

hbase海量实时存储（百万/秒，数据库千/秒），druid(万亿/秒)。hbase经常使用于海量数据直接存储或分析后数据实时查询等，druid常做为流数据计算后实时查询。本文将介绍两者的架构/分布式部署，存储结构/格式，对比等。

hbase

官方：Apache HBase™ is the Hadoop database, a distributed, scalable, big data store.
Use Apache HBase™ when you need random,realtime read/write access to your Big Data. 硬件成本不高状况下托管数十亿行*百万列。开发思路基于bigtable。文件系统基于hdfs（这部分分布式相关的一致性可用性再也不说了）

架构

CF——region(只有一个CF)s——多个stores——一个memstore+多个HFILE

• HBase Master
  用于协调多个 Region Server，侦测各个 Region Server 之间的状态，并平衡 Region Server 之间的负载。HBase Master 还有一个职责就是负责分配 Region 给 Region Server。
• zk
  除了master的管理，元数据路由也放在zk中保证该HA，client先和zk通讯。
• Region Server
  每个 Region Server 管理着不少个 Region。对于 HBase 来讲，Region 是 HBase 并行化的基本单元。所以，数据也都存储在 Region 中。
• Region 分区（m_p基本单元）
  所能存储的数据大小是有上限的，当达到该上限时（Threshold），Region 会进行分裂，数据也会分裂到多个 Region 中，这样即可以提升数据的并行化，以及提升数据的容量。每一个server的region数量不易过多，缘由分区占内存，每次满每次都会刷新，map_reduce时数据量（100比较合适）见：http://hbase.apache.org/book....
• Store/MemStore
  每一个 Region 包含着多个 Store 对象。每一个store是一个CF数据，每一个 Store 包含一个 MemStore，和一个或多个 HFile。
  MemStore 即是数据在内存中的实体，而且通常都是有序的。当数据向 Region 写入的时候，会先写入 MemStore。当 MemStore 中的数据须要向底层文件系统倾倒（Dump）时（例如 MemStore 中的数据体积到达 MemStore 配置的最大值），Store 便会建立 StoreFile.

• StoreFile/Hfile
  而 StoreFile 就是对 HFile 一层封装。因此 MemStore 中的数据会最终写入到 HFile 中，也就是磁盘 IO。因为 HBase 底层依靠 HDFS，所以 HFile 都存储在 HDFS 之中。

  Table                    (HBase table)
      Region               (Regions for the table)
          Store            (Store per ColumnFamily for each Region for the table)

高可用

• master
  多副本，zk自动故障转移

• region
  通常的region server不可用，须要zk的发现-从新分配-恢复，期间该分区不可用，对于不能容忍的新版实现：region server多副本：https://hbase.apache.org/book...
  region自动分片和region server自动故障转移(master检测后，从新分配region,更新zk的meta)，WAL 保存在HDFS 的 /hbase/.logs/ 里面，每一个region一个文件。
  region server会在mem flush时把hfile压缩为更少，更大的文件，检测该区域的hfile文件过大会自动拆分，
  region从新分配过程：

  

  1.RegionServer在本地决定拆分区域，并准备拆分。获取共享读锁，以防止在拆分过程当中修改模式。zookeeper下建立一个znode /hbase/region-in-transition/region-name，并将znode的状态设置为SPLITTING。
  2.Master watcher获取
  3.RegionServer建立一个在HDFS中.splits父region目录下命名的子目录。
  4.RegionServer关闭父区域，并在其本地数据结构中将该区域标记为脱机。该region离线。此时，将发送到父区域的客户端请求NotServingRegionException。客户端将重试。
  5.RegionServer在目录下.splits为子区域A和B建立区域目录，并建立必要的数据结构。而后分割存储文件，建立两个引用到父区域  
  6.RegionServer在HDFS中建立实际的区域目录，并移动每一个子项的引用。
  
  7.RegionServer向表发送Put请求.META.，将父级设置为脱机，并添加有关子区域的信息。成功以后才有单独的条目，不然清理回滚
  8.RegionServer并行打开A和B.
  9.RegionServer将女儿A和B .META.以及它托管区域的信息添加到其中。分裂地区（带有父母参考的孩子）如今在线。在此以后，客户能够发现新区域并向其发出请求。客户端在.META.本地缓存条目，可是当它们向RegionServer发出请求时.META.，它们的缓存将失效，而且它们将从中了解新区域.META.。 
  10.RegionServer将/hbase/region-in-transition/region-nameZooKeeper中的znode更新为state SPLIT，以便master能够了解它。若有必要，平衡器能够自由地将子区域从新分配给其余区域服务器。分裂交易现已完成。 
  11.拆分后，.META.HDFS仍将包含对父区域的引用。当子区域中的压缩重写数据文件时，将删除这些引用。主服务器中的垃圾收集任务会按期检查子区域是否仍然引用父区域的文件。若是不是，则将删除父区域。

• WAL
  一个region server一个WAL（串行写入HDFS瓶颈，multiwal在底层HDFS中使用多个管道并行写入，按区域划分你传入，不研究了）
  打开某个区域时，须要重播属于该区域的WAL文件中的编辑。所以，WAL文件中的编辑必须按区域分组，以即可以重放特定的集合以从新生成特定区域中的数据。按区域对WAL编辑进行分组的过程称为日志分割。
  旧的日志切割任务由hmaster完成并由zk协调，region server多时压力大=》改成每一个region server执行

  1.重命名WAL  / HBase的/ WALS / <主机>，<端口>，<起始码> -splittin
  2.日志分割器一次一个读取日志文件，放入region缓存中，启动多个写线程，每一个读region缓存到临时文件/hbase/<table_name>/<region_id>/recovered.edits/.temp。分割后.temp文件重命名为sequennce id做为第一个。
  3.日志拆分完成后，每一个受影响的区域都将分配给RegionServer。
  当区域被打开，recovered.edits文件夹中检查恢复的编辑文件。若是存在任何此类文件，则经过阅读编辑并将其保存到MemStore来重放它们。重放全部编辑文件后，MemStore的内容将写入磁盘（HFile）并删除编辑文件。

• hfile
  hdfs的三副本等实现。

数据格式

基于列存储
HBase 中建立表格时，就须要指定表格的 CF、Row-key 以及 Qulifier
Row-key 加上 CF 加上 Qulifier 再加上一个时间戳才能够定位到一个单元格数据（Hbase 中每一个单元格默认有 3 个时间戳的版本数据）
列与行之间的转化如图：

http://cloudepr.blogspot.com/...

HFILE

HFILE相似Google’s SSTable，不分层
内存占用：key不宜过大（全部index放内存key长度*n）,1G占用1.2M（15600 (1GB/64KB)*(64+)）
The trailer, file-info and total data block indexes (optionally, may add meta block indexes)

• HFile压缩
  Gzip,LZO(低压缩率高解压速度)snappy(代替lzo),LZ4
  long keys（比数据长）/列多：前缀编码FAST_DIFF
  值大：块压缩
  冷数据：GZIP，压缩率高
  热数据：snappy/lzo
• hfile合并（指的minor合并，合并部分小的相邻的，Major是同一个store中的合并成一个）
  ExploringCompaction：合并小文件到大文件
  DateTieredCompactionPolicy：时间戳分层，1.3后
  StripeCompactionPolicy：相似leveldb的两层，Major Compaction只到level0。minor包含两步：level0到level1的合并，level1按照ExploringCompaction合并
  https://juejin.im/post/5bfe97...

事务

• 一个region中
• 写
  对行（一行或多行）加锁，屏蔽对相同行的并发写；
  获取当前的 WriteNumber；
  提交修改到 WAL；
  提交修改到 Memstore（用前面获取的 WriteNumber 标记修改的 KeyValues）；
  提交事务，也就是把 ReadPoint 更新为当前获取的 WriteNumber；
  释放行（一行或多行）锁。
• scan:
  打开 scanner；
  获取当前的 ReadPoint；
  用获取的 ReadPoint 过滤全部扫描到的 KeyValues（KeyValues 的 memstore timestamp > ReadPoint，只看 ReadPoint 以前的）；(防止compact,delete读不完整行)。
  关闭 scanner（scanner 由客户端初始化）。
• 顺序保证：一个事务的提交会被推迟，直到先前的事务提交
• 持久：刷内存或盘盘

与RDBMS对比

只用于数据量大的，HDFS 默认会将每个 Block 数据备份 3 分，硬件仍是比较多

druid

Druid的重点是极低延迟查询，Druid彻底索引全部数据。
高性能，亚秒响应的交互式查询（万亿行数据集上执行 query，响应时间小于1秒）；可扩展性，采用分布式shared-nothing的架构，能够扩展到PB级；支持聚合函数，count和sum，近似查询的Aggregator。只支持流失插入，不支持流失更新，不支持大表join。适用于插入高更新少，有时间，列式聚合和报告等。较低的成本（比ES在聚合的成本低）
关键点：列存储，倒排索引，Roll UP，roaring 或者concise bitmap位图索引以及高效的压缩支撑了Druid的高效查询
常做用与spark之上。
官方：http://druid.io/docs/latest/d...
为每一个定向类别（包括日期）建立一个单独的column family，Hbase受限于rowkey设计，不能很好地解决多维分析。另外Hbase自己没有为column family建立bitmap indexing，查询速度应该会受到影响。

架构：

• Coordinator web (ui管理群集，下发元数据给zk,不直接交互)。
• Overlord 控制数据提取工做负载的分配(middlermanage-overload-具体middlemanager)，协调任务建立锁。
• broker 处理来自外部客户端的查询，从zk和获取路由，段缓存，合并结果，历史会在zk中注册本身的段。
• router 可选的，能够将请求路由到Broker，Coordinator和Overlords。
• 历史
  存储可查询数据，coordinator经过在zk的load queue目录下建立实例分配新的段给历史，当历史发现，先查本地缓存配置信息，没有从zk下载元数据，包含deep storage调的位置和解压方式，完成后通知zk修改段信息供查询。下载数据放入内存
• MiddleManager 运行任务。

外部依赖

• Mysql（metadata storage）
  存储Druid中的各类metadata（里面的数据都是Druid自身建立和插入的），包含3张表：”druid_config”（一般是空的）, “druid_rules”（coordinator nodes使用的一些规则信息，好比哪一个segment从哪一个node去load）和“druid_segments”（存储每一个segment的metadata信息）；
• Deep storage
  存储segments，Druid目前已经支持本地磁盘，NFS挂载磁盘，HDFS，S3等。Deep Storage的数据有2个来源，一个是batch Ingestion, 另外一个是real-time nodes；
• ZooKeeper
  被Druid用于管理当前cluster的状态，好比记录哪些segments从Real-time nodes移到了Historical nodes；

数据组织

datasources->chunck(按时间分区)->segment

segment的数据生成过程

• MiddleManager

1. MiddleManager上建立
   若是是append模式，overload分配partition添加到现有段，覆盖模式锁定（时间间隔，chunck的范围）建立一个新版本段。
   实时任务，可当即查询该段，但未发布
   在读取过程当中会：
   转为列
   建立位图索引
   压缩
2. 读完写入deep storage（多副本） =>发布：写入metadata stroage( schema of the segment, its size, and its location on deep storage)
   若是实时任务，等待历史进程加载该段，不然当即退出

• 对于core/historical的过程:
  协调器按期轮询元数据存储（默认状况下，每1分钟一次），用于新发布的段。
  当协调器找到已发布和使用但不可用的段时，它会选择历史进程来加载该段并指示历史记录执行此操做。
  历史记录加载段并开始提供它。
  若是任务正在等待切换（发布由his提供服务），则它将退出。

不支持单键更新
历史数据根据时间分层，冷/热
段能够继续分片和合并

查询过程

1.broker识别数据再哪些Historicals和MiddleManagers，发送子查询。
其中为了减小查询：
1）.broker的purge修剪：找段
2）.每段用索引，行过滤器找行（bitmap）
3）.查询所需列时，能够在行之间跳过。见特性
2.处理后broker合并

特性

• Rollup会将采用其设定的聚合器进行聚合
• 列三种类型：时间，列，统计
  1.时间戳和统计：变长整数编码+LZ4压缩的整数或浮点
  2.列：字典+列字典值+倒排索引
  1）将值（老是被视为字符串）映射到整数ID的字典，
  2）列的值列表，使用1中的字典编码，和
  3）对于列中的每一个不一样值，一个位图指示哪些行包含该值。倒排索引：posting list采用压缩的bitmap位图索引，BitMap支持Consice和Roaring两种压缩编码方式

  
  图中标识第一列的索引

• 布尔查询

  

  以这个SQL查询为例，select sum(click) from table where time between 2016-05-24T11 and 2016-05-24T12 and广告=C2 and 地域=P1;首先根据时间段定位到Segment，而后根据广告=C2和地域=P1，获得 他们各自的字典编码，C2=1，P1=0，而后根据字典编码获得BitMap的Offet，从而获得Bitmap，C2=1 index为1的bitmap为0110，同理获得P1的bitmap为1100，0110和1100进行And与运算是0100，获得的offset是1，说明咱们须要在click中读取offset=1的值是4.