[大数据]HBase之异步Replication机制

异步Replication概述

HBase 提供了一种群集复制机制，允许使用源群集的预写日志（WAL）来传播更改，从而使一个群集的状态与另一个群集的状态保持同步。集群复制的一些用例包括：

• 备份和灾难恢复
• 数据聚合
• 地理数据分布
• 在线数据提取与离线数据分析相结合
  以列族的粒度启用复制。在为列族启用复制之前，请在目标群集上创建要复制的表和所有列族。复制是异步的，因为我们将 WAL 发送到后台的另一个集群，这意味着当您想通过复制进行恢复时，可能会丢失一些数据。

异步Replication原理

HBase中的Replication是基于WAL的，其在主集群的每一个RegionServer进程内部起了一个叫作ReplicationSource的线程来负责Replication，同时在备集群的每一个RegionServer内部起了一个ReplicationSink的线程来负责接收Replication数据。ReplicationSource记录须要同步的WAL队列，而后不断读取WAL中的内容，同时能够根据Replication的配置作一些过滤，好比是否要复制这个表的数据等，而后经过replicateWALEntry这个Rpc调用来发送给备集群的RegionServer，备集群的ReplicationSink线程则负责将收到的数据转换为put/delete操做，以batch的形式写入到备集群中。
由于是后台线程异步的读取WAL并复制到备集群，因此这种Replication方式叫作异步Replication，正常状况下备集群收到最新写入数据的延迟在秒级别。

WAL Edit 生命周期

单个 WAL 编辑将经历几个步骤，以便复制到从属群集。

1. HBase 客户端使用 Put 或 Delete 操作来操作 HBase 中的数据。
2. RegionServer以某种方式将请求写入 WAL，如果未成功写入则允许重放。
3. 如果更改的单元格对应于作为复制作用域的列族，则会将编辑添加到队列以进行复制。
4. 在单独的线程中，作为批处理的一部分，从日志中读取编辑。仅保留符合复制条件的 KeyValues。可复制的 KeyValues 是列族的一部分，其列的范围为 GLOBAL，不是hbase:meta等目录的一部分，不是源自目标从属群集，并且尚未被目标从群集使用。
5. 编辑用主 UUID 标记并添加到缓冲区。填充缓冲区或读取器到达文件末尾时，缓冲区将发送到从属群集上的随机区域服务器。
6. RegionServer按顺序读取编辑并将它们分成缓冲区，每个表一个缓冲区。读取所有编辑后，使用 HBase 的普通客户端表刷新每个缓冲区。主服务器的 UUID 和已经使用数据的从服务器的 UUID 将保留在应用的编辑中，以防止复制循环。
7. 在主服务器中，当前正在复制的 WAL 的偏移量在 ZooKeeper 中注册。
8. 插入编辑的前三个步骤是相同的。
9. 同样在单独的线程中，RegionServer以与上面相同的方式读取，过滤和编辑日志编辑。从属RegionServer不响应 RPC 调用。
10. 主库睡眠并再次尝试可配置的次数。
11. 如果从属RegionServer仍然不可用，则主服务器选择要复制到的RegionServer的新子集，并再次尝试发送编辑缓冲区。
12. 同时，WALs 被滚动并存储在 ZooKeeper 的队列中。由RegionServer将归档的日志从RegionServer的日志目录移动到中央日志目录，将更新其在复制线程的内存中队列中的路径。
13. 当从属群集最终可用时，缓冲区的应用方式与正常处理时相同。然后，主RegionServer将复制在中断期间累积的积压日志。

相关命令

• add_peer
  在两个群集之间添加复制关系。
  o ID - 唯一字符串，不得包含连字符。
  o CLUSTER_KEY：使用以下模板和适当的占位符组成：hbase.zookeeper.quorum:hbase.zookeeper.property.clientPort:zookeeper.znode.parent。可以在主 UI 信息页面上找到此值。
  o STATE（可选）：ENABLED 或 DISABLED，默认值为 ENABLED
• list_peers
  列出此群集已知的所有复制关系
• enable_peer
  启用先前禁用的复制关系
• disable_peer
  禁用复制关系。 HBase 将不再向该对等集群发送编辑，但它仍会跟踪在重新启用时需要复制的所有新 WAL。只要存在对等体，在启用或禁用复制时将保留 WAL。
• remove_peer
  禁用并删除复制关系。 HBase 将不再向该对等集群发送编辑或跟踪 WAL。
• enable_table_replication <table_name>
  为其所有列系列启用表复制开关。如果在目标集群中找不到该表，则它将创建一个具有相同名称和列族的表。
• disable_table_replication <table_name>
  禁用其所有列系列的表复制开关。

参数调优

• replication.source.ratio，默认值为0.1。这样导致slave集群里只有10%对外提供转发服务。如果客户的从集群完全用于容灾没有读请求，可以考虑更改成1。
• hbase.replication.source.maxthreads，默认值是10。增大可以提高主从replication的吞吐量。
• replication.source.size.capacity，默认是64M。在replication时一次复制entry包多大数据。和参数replication.source.nb.capacity一起使用，表示最多发多少个entry包，默认是25000。一次太多可能会有问题。可参考Hbase使用的是2M，2000个。

管理并配置集群复制

以下步骤具体描述如何配置两个集群之间的Binlog复制功能。

1. HBase默认此特性是关闭的，需要在源集群上进行设定并重启集群，将hbase.replication参数设定为true
   

2. 在源集群修改replication.sleep.before.failover配置设置为1，此配置需配置在hbase-site.xml 的 HBase 服务高级配置代码段（安全阀）中添加
   
   

3. 在源集群上添加两个集群的复制关系，在hbase shell中执行命令。下面的命令指明了目标集群的Zookeeper地址，以及它在这个Zookeeper上路径

hbase shell > add_peer '1',"zknode1,zknode2,zknode3:2181:/hbase"

有关add_peer的详细用法，请参考HBase官方文档说明https://hbase.apache.org/book.html#_cluster_replication

4. 在源集群的表上标注出需要备份的列族信息，可以使用以下命令来针对整个表配置复制关系

hbase shell > enable_table_replication 'TABL'

注：如果不想再对表进行复制，可以使用以下命令来禁用表的复制

hbase shell > 
disable_table_replication 'TABL'

5. 检查日志查看是否正在进行复制，查看RegionServer日志相关日志如下
   LOG.info("Replicating "+clusterId + " -> " + peerClusterId);

监控工具

VerifyReplication

VerifyReplication MapReduce 作业（包含在 HBase 中）对两个不同集群之间的复制数据进行系统比较。在主群集上运行 VerifyReplication 作业，为其提供用于验证的对等 ID 和表名。您可以通过指定时间范围或特定系列来进一步限制验证。这份工作的简称是verifyrep。要运行作业，请使用如下命令：

hbase org.apache.hadoop.hbase.mapreduce.replication.VerifyReplication -mappers 10 -bandwidth 1024 1 'TRAF_RSRVD_5:TRAFODION._DTM_.TRAFODION_BINLOG'

status replication

您可以使用 HBase Shell 命令status 'replication’来监视群集上的复制状态。该命令有三个变体：

• status ‘replication’ - 打印每个源及其接收器的状态，按主机名排序。
• status ‘replication’, ‘source’ - 打印每个复制源的状态，按主机名排序。
• status ‘replication’, ‘sink’ - 打印每个复制接收器的状态，按主机名排序。
  通常只需要执行 status ‘replication’ 即可。这个命令会显示出live的RegionServer，并显示这些RegionServer当前的传输状态。
• 在源端（SOURCE）：
  PeerID : 复制连接id;
  AgeOfLastShippedOp : 上一次复制数据给Slave的age(这个指标更详细说明请参考repliation源码分析);
  SizeOfLogQueue : 当前复制队列的HLog数量;
  TimeStampsOfLastShippedOp : 上一次复制数据给Slave的时间戳;
  Replication Lag : 复制延迟时间
• 在目的端（SINK）：
  AgeOfLastAppliedOp: 上次复制的age;
  TimeStampsOfLastAppliedOp：上次获得的时间戳

示例输出：

hbase(main):009:0> status 'replication'
version 1.2.0-cdh5.16.2
4 live servers
    gyw29.esgyn.cn:
       SOURCE: PeerID=1, AgeOfLastShippedOp=3973899, SizeOfLogQueue=48, TimeStampsOfLastShippedOp=Mon Jun 27 09:28:46 CST 2022, Replication Lag=3977761
       SINK  : AgeOfLastAppliedOp=0, TimeStampsOfLastAppliedOp=Mon Jun 27 09:16:59 CST 2022
    gyw32.esgyn.cn:
       SOURCE: PeerID=1, AgeOfLastShippedOp=3909927, SizeOfLogQueue=55, TimeStampsOfLastShippedOp=Mon Jun 27 09:28:46 CST 2022, Replication Lag=3979522
       SINK  : AgeOfLastAppliedOp=0, TimeStampsOfLastAppliedOp=Mon Jun 27 09:16:59 CST 2022
    gyw30.esgyn.cn:
       SOURCE: PeerID=1, AgeOfLastShippedOp=3922388, SizeOfLogQueue=50, TimeStampsOfLastShippedOp=Mon Jun 27 09:28:46 CST 2022, Replication Lag=3979922
       SINK  : AgeOfLastAppliedOp=0, TimeStampsOfLastAppliedOp=Mon Jun 27 09:16:59 CST 2022
    gyw31.esgyn.cn:
       SOURCE: PeerID=1, AgeOfLastShippedOp=3954983, SizeOfLogQueue=59, TimeStampsOfLastShippedOp=Mon Jun 27 09:28:47 CST 2022, Replication Lag=3976921
       SINK  : AgeOfLastAppliedOp=0, TimeStampsOfLastAppliedOp=Mon Jun 27 09:16:59 CST 2022

HBase jmx

以使用HBase API来获取更详细的Replication相关的数据，命令如下

curl http://<RegionServer_IP>:60030/jmx?qry=Hadoop:service=HBase,name=RegionServer,sub=Replication

以上命令显示某个指定的RegionServer的Replication信息。
样例输出：

{
  "beans" : [ {
    "name" : "Hadoop:service=HBase,name=RegionServer,sub=Replication",
    "modelerType" : "RegionServer,sub=Replication",
    "tag.Context" : "regionserver",
    "tag.Hostname" : "gyw29.esgyn.cn",
    "source.1.logEditsRead" : 941,
    "source.1.logReadInBytes" : 422045,
    "source.ageOfLastShippedOp" : 3355072,
    "source.1.shippedKBs" : 0,
    "source.1.ignoredUncleanlyClosedLogContentsInBytes" : 0,
    "source.logReadInBytes" : 1045079,
    "source.1.sizeOfLogQueue" : 43,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.completedLogs" : 0,
    "source.sizeOfLogQueue" : 43,
    "sink.ageOfLastAppliedOp" : 0,
    "source.1.ageOfLastShippedOp" : 3355072,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.logEditsFiltered" : 891,
    "source.shippedOps" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.logEditsRead" : 891,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.completedRecoverQueues" : 1,
    "source.1.closedLogsWithUnknownFileLength" : 0,
    "source.1.repeatedLogFileBytes" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.sizeOfLogQueue" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.shippedBatches" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.shippedOps" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.restartedLogReading" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.sizeOfLogQueue" : 0,
    "sink.appliedOps" : 0,
    "source.1.shippedOps" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.shippedBatches" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.ageOfLastShippedOp" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.shippedKBs" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.uncleanlyClosedLogs" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.ignoredUncleanlyClosedLogContentsInBytes" : 0,
    "source.1.logEditsFiltered" : 940,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.repeatedLogFileBytes" : 0,
    "source.1.shippedBatches" : 0,
    "source.1.completedRecoverQueues" : 0,
    "source.1.completedLogs" : 0,
    "sink.appliedBatches" : 0,
    "source.uncleanlyClosedLogs" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.uncleanlyClosedLogs" : 0,
    "source.closedLogsWithUnknownFileLength" : 0,
    "source.repeatedLogFileBytes" : 0,
    "source.shippedKBs" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.logEditsRead" : 890,
    "source.1.restartedLogReading" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.repeatedLogFileBytes" : 0,
    "source.completedRecoverQueues" : 2,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.shippedKBs" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.closedLogsWithUnknownFileLength" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.completedRecoverQueues" : 1,
    "source.restartedLogReading" : 0,
    "source.ignoredUncleanlyClosedLogContentsInBytes" : 0,
    "source.logEditsRead" : 2722,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.ageOfLastShippedOp" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.logEditsFiltered" : 890,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.shippedOps" : 0,
    "source.1.uncleanlyClosedLogs" : 0,
    "source.logEditsFiltered" : 2721,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.completedLogs" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.logReadInBytes" : 311523,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.closedLogsWithUnknownFileLength" : 0,
    "source.1-gyw31.esgyn.cn,60020,1656292333815.ignoredUncleanlyClosedLogContentsInBytes" : 0,
    "source.completedLogs" : 0,
    "source.shippedBatches" : 0,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.logReadInBytes" : 311511,
    "source.1-gyw30.esgyn.cn,60020,1656292333740.restartedLogReading" : 0
  } ]
}