前言

笔者之前写过一篇关于存储系统数据探查分析的文章：存储系统“数据之眼”的设计–数据探查服务。文章主要阐述了关于对存储系统，我们如何对其上存储的海量数据做一个常规分析同时不能影响系统提供正常的数据服务。其中的一个核心思想是构建一个探查服务，这个服务定期获取存储系统db文件快照，然后做二次分析，并将结果存入到结果汇聚表中。本文笔者将继续聊聊目前社区对此的进一步优化实现：能够基于RocksDB的Delta Update的方式做更加准实时地结果分析服务。

Ozone数据探查服务：Recon Server

为了使大家能够了解本文笔者将要阐述的，这里有必要提及Ozone数据探查服务Recon Server的服务设计架构，如下图所示：

在这里插入图片描述

在早些时间，社区已经实现了从OM Follower到Recon Server的Checkpoint for bootstrap的分支。目前下面的分支apply WAL update也即将要实现，这个过程也是笔者本文将要阐述的主要内容：Recon Server基于RocksDB的WAL做Delta更新。

Recon Server基于Checkpoint获取定期DB Snapshot的弊端

在Recon Server最开始实现RocksDB同步的时候，一开始了采用的是周期性同步DB文件的办法。这个DB来自于Ozone OzoneManager Follower节点的RocksDB文件。相当于是去定期去拉去OM Follower节点DB的snapshot版本。而这个snapshot版本由RocksDB原生支持的Checkpoint行为产生。

这种方法虽然说比较简单，直接，但是它并不是最高效和完美的，至少它有以下一些问题：

无法做到准实时同步，数据的实效性完全取自于Recon Server服务配置的同步间隔时间。
如果为了增强数据的实时性，增大中期拉取快照频率，那RocksDB频繁的checkpoint行为势必会有一定的开销。而且Checkpoint出的文件在bootstrap过程中，会经过压缩、传输、再解压缩到Recon Server机器上。整个过程步骤繁多，解压缩文件的过程也将是一个耗时的操作。

因此在这里，一个更优的策略方式应该是：整体基于Delta的DB更新策略，Checkpoint文件做定期的整体数据校准。

RocksDB基于Update WAL的Delta更新

Ozone基于的是RocksDB做元数据DB的存储。那么我们不禁在想，在RocksDB中我们是否能够得到距离某次transaction提交后的增量更新呢？例如Update WAL这类的信息？答案是有的，RocksDB的内部API getUpdatesSince(final long sequenceNumber) 能够很好地帮助我们解决这个问题。

  /**
   * <p>Returns an iterator that is positioned at a write-batch containing
   * seq_number. If the sequence number is non existent, it returns an iterator
   * at the first available seq_no after the requested seq_no.</p>
   *
   * <p>Must set WAL_ttl_seconds or WAL_size_limit_MB to large values to
   * use this api, else the WAL files will get
   * cleared aggressively and the iterator might keep getting invalid before
   * an update is read.</p>
   *
   * @param sequenceNumber sequence number offset
   *
   * @return {@link org.rocksdb.TransactionLogIterator} instance.
   *
   * @throws org.rocksdb.RocksDBException if iterator cannot be retrieved
   *     from native-side.
   */
  public TransactionLogIterator getUpdatesSince(final long sequenceNumber)
      throws RocksDBException {
    return new TransactionLogIterator(
        getUpdatesSince(nativeHandle_, sequenceNumber));
  }

这里的参数sequenceNumber可以理解为是transaction id。Sequence number会在每次的RocksDB写记录中进行递增，不过它不是针对单独的一条写记录，而是一个批的写，在RocksDB中的概念叫做WriteBatch。一个WriteBatch中可能包含一条记录，也可能有多个记录包含在此批中，等待被写入。

在RocksDB的增量更新中，我们就能够有效地利用这个API。每次我们在apply完最新一批的transaction之后，更新当前同步的sequence number，以便下次查询时使用。

在拿到Update的WAL之后，Recon Server做了以下的更新事务的处理

1）对查询得到的byte数组数据进行transaction对象的转换，这里需要实现一个WriteBatch.Handler的子类来监听RocksDB的update event并执行处理逻辑。Recon在这里实现了对应的Handler实现类OMDBUpdatesHandler。
2）OMDBUpdatesHandler将解析完的OMDBUpdateEvent列表传入ReconDBUpdate task中进行处理。
3）ReconDBUpdate task内部根据transaction信息进行SQL汇聚表的结果更新。这里的Task能够选择针对特定表的事务更新。Task Update Task采用异步执行的方式运行于统一线程池内，中间如若出现部分Handler执行出错的情况，则进行重试执行。

上述过程的流程图如下所示：
在这里插入图片描述

以上就是Ozone数据探查服务Recon Server内部基于OM Follower DB的Delta更新过程了，相信能够给大家带来借鉴意义。

引用

[1].https://issues.apache.org/jira/browse/HDDS-1084. Ozone Recon Service
[2].https://blog.csdn.net/Androidlushangderen/article/details/88783397