精华内容
下载资源
问答
  • HBase预写日志WAL机制

    千次阅读 2019-03-10 17:28:17
    预写日志(Write-ahead log,WAL) 最重要的作用是灾难恢复,一旦服务器崩溃,通过重放log,我们可以恢复崩溃之前的数据。如果写入WAL失败,整个操作也将认为失败。 从上图看: 1 客户端对数据执行一个修改操作,如...

    预写日志(Write-ahead log,WAL)
    最重要的作用是灾难恢复,一旦服务器崩溃,通过重放log,我们可以恢复崩溃之前的数据。如果写入WAL失败,整个操作也将认为失败。

    WAL

    从上图看:
    1 客户端对数据执行一个修改操作,如put(),delete(),incr()等。
    2 每一个修改被封装到一个KeyValue对象实例,并通过RPC调用发送出来。
    3 上述调用成批地发送给含有匹配region的HRegionServer。
    4 数据先被写入到WAL,然后被放放到实际拥有记录的存储文件的MemStore中。
    5 当MemStore达到一定的大小或经历一个特定时间之后,数据会异步地连续写入到文件系统中。

    由于实际的日志存储在HDFS上,所以即使在服务器完全崩溃的情况下,WAL也能保证数据不会丢失。其他服务器可以打开日志文件然后回放这些修改。

    为什么要一个RegionServer 对应于一个HLog。为什么不是一个region对应于一个log file?
    引用BigTable中的一段话:
    如果我们每一个“tablet”(对应于HBase的region)都提交一个日志文件,会需要并发写入大量的文件到GFS,这样,根据每个GFS
    server所依赖的文件系统,写入不同的日志文件会造成大量的磁盘操作。
    HBase依照这样的原则。在日志被回滚和安全删除之前,将会有大量的文件。如果改成一个region对应于一个文件,将会不好扩展,迟早会引发问题。

    延迟(异步)同步写入WAL

    WAL在默认情况下时开启的,当然,我们也可以手动关闭。调用{Mutation.setDurability(Durability.SKIP_WAL)}方法来关闭,这样做的确可以使得数据操作快一点,但并不建议这样做,一旦服务器宕机,数据就会丢失。

    延迟(异步)同步写入WAL。调用setDurability(Durability.ASYNC_WAL),这样通过设置时间间隔来延迟将操作写入WAL。
    时间间隔:HBase间隔多久会将操作从内存写入到WAL,默认值为1s。 这种方法也可以相对应地提高性能。

    WAL几个重要的类
    1、HLog

    HLog是实现WAL的类。一个HRegionServer对应一个HLog实例。当HRegion初始化时,HLog将作为一个参数传给HRegion的构造函数。
    HLog最核心的是调用doWrite的append()方法,前面提到的可能对数据改动的操作都就将首先调用这个方法。出于性能的考虑,put(), delete()和incrementColumnValue()有一个开关函数setWriteToWAL(boolean) ,
    设为false将禁用WAL。这是为什么上图中向下的箭头是虚线的原因。默认时候当然需要WAL,但是假如你运行一个数据导入的MapReduceJob,你可以通过关闭WAL获得性能上的提升。

    2、HLogKey类

    1、当前的WAL使用的是hadoop的sequencefile格式,其key是HLogKey实例。HLogKey中记录了写入数据的归属信息,,除了table和region名字外,同时还包括sequencenumber和timestamp,timestamp是“写入时间“,sequencenumber的起始值为0,或者是最近一次存入文件系统中sequence number
    2、HLog sequenceFile的value是HBase的KeyValue对象,即对应HFile中的KeyValue

    3、WALEdit类

    客户端发送的每个修改都会封装成WALEdit类,一个WALEdit类包含了多个更新操作,可以说一个WALEdit就是一个原子操作,包含若干个操作的集合

    4、LogSyncer类

    Table在创建的时候,有一个参数可以设置,是否每次写Log日志都需要往集群里的其他机器同步一次,默认是每次都同步,同步的开销是比较大的,但不及时同步又可能因为机器宕而丢日志。

    同步的操作现在是通过Pipeline的方式来实现的,Pipeline是指datanode接收数据后,再传给另外一台datanode,是一种串行的方式;
    n-Way Writes是指多datanode同时接收数据,最慢的一台结束就是整个结束。
    差别在于一个延迟大,一个并发高,hdfs现在正在开发中,以便可以选择是按Pipeline还是n-Way Writes来实现写操作。

    Table如果设置每次不同步,则写操作会被RegionServe缓存,并启动一个LogSyncer线程来定时同步日志,定时时间默认是1秒,也可由hbase.regionserver.optionallogflushinterval设置。

    5、LogRoller类

    日志写入的大小是有限制的。LogRoller类会作为一个后台线程运行,在特定的时间间隔内滚动日志。通过hbase.regionserver.logroll.period属性控制,默认1小时。

    WAL滚动
    就是上文提到的LogRoller类。
    通过wal日志切换,这样可以避免产生单独的过大的wal日志文件,这样可以方便后续的日志清理(可以将过期日志文件直接删除)另外如果需要使用日志进行恢复时,也可以同时解析多个小的日志文件,缩短恢复所需时间。
    触发滚动的条件:

    1.SyncRunner线程在处理日志同步后, 检查当前在写的wal的日志大小是否超过配置{hbase.regionserver.hlog.blocksize默认为hdfs目录块大小}*{hbase.regionserver.logroll.multiplier默认0.95},超过后同样调用requestLogRoll发起日志滚动请求
    2.SyncRunner线程在处理日志同步后,如果有异常发生,就会调用requestLogRoll发起日志滚动请求

    WAL归档和删除
    归档:WAL创建出来的文件都会放在/hbase/.log下,在WAL文件被定为归档时,文件会被移动到/hbase/.oldlogs下
    删除:1.判断:是否此WAL文件不再需要,是否没有被其他引用指向这个WAL文件

    会引用此文件的服务:
    (1)TTL进程:该进程会保证WAL文件一直存活直到达到hbase.master.logcleaner.ttl定义的超时时间(默认10分钟)为止
    (2)(2)备份(replication)机制:如果你开启了HBase的备份机制,那么HBase要保证备份集群已经完全不需要这个WAL文件了,才会删除这个WAL文件。这里提到的replication不是文件的备份数,而是0.90版本加入的特性,这个特性用于把一个集群的数据实时备份到另外一个集群。如果你的手头就一个集群,可以不用考虑这个因素。

    2.删除

    Write-Ahead-Log(WAL)保证数据的高可用性。
    如果没有 WAL,当RegionServer宕掉的时候,MemStore 还没有写入到HFile,或者StoreFile还没有保存,数据就会丢失。
    HBase中的HLog机制是WAL的一种实现,每个RegionServer中都会有一个HLog的实例,RegionServer会将更新操作(如 Put,Delete)先记录到 WAL(也就是HLog)中,然后将其写入到Store的MemStore,最终MemStore会将数据写入到持久化的HFile中(MemStore 到达配置的内存阀值)。这样就保证了HBase的高可用性。

    下一章:HBase安装部署

    展开全文
  • HLog失效:写入数据一旦从MemStore中刷新到磁盘,HLog(默认存储目录在/hbase/WALs下)就会自动把数据移动到 /hbase/oldWALs 目录下,此时并不会删除 Hlog删除:Master启动时会启动一个线程,定期去检查oldWALs目录...

    Hlog WALs和oldWALs

    这里先介绍一下Hlog失效和Hlog删除的规则

    HLog失效:写入数据一旦从MemStore中刷新到磁盘,HLog(默认存储目录在/hbase/WALs下)就会自动把数据移动到 /hbase/oldWALs 目录下,此时并不会删除

    Hlog删除:Master启动时会启动一个线程,定期去检查oldWALs目录下的可删除文件进行删除,定期检查时间为 hbase.master.cleaner.interval ,默认是1分钟 ,删除条件有两个:

    1.Hlog文件在参与主从复制,否的话删除,是的话不删除
    2.Hlog文件是否在目录中存在 hbase.master.logcleaner.ttl 时间,如果是则删除
    

    整体流程

    pos 格式流程图下载地址:

    链接:https://pan.baidu.com/s/1szhpVn7RyegE0yqQedACIA 
    提取码:ig9x
    

    img

    这里只介绍与wal相关的流程,一下介绍的代码都在上图中标记类名,方法名,以及说明,可以直接从源码中查看

    HMaster 初始化

    HMaster启动初始化 ,HMaster构造方法调用 startActiveMasterManager 方法

    startActiveMasterManager 方法 调用 finishActiveMasterInitialization(status); 方法

    在 finishActiveMasterInitialization 方法中会启动所有服务线程,代码段如下

    // start up all service threads.
    status.setStatus("Initializing master service threads");
    startServiceThreads();
    

    startServiceThreads 方法代码如下,

     /*
       * Start up all services. If any of these threads gets an unhandled exception
       * then they just die with a logged message.  This should be fine because
       * in general, we do not expect the master to get such unhandled exceptions
       *  as OOMEs; it should be lightly loaded. See what HRegionServer does if
       *  need to install an unexpected exception handler.
       */
      private void startServiceThreads() throws IOException{
       // Start the executor service pools
       this.service.startExecutorService(ExecutorType.MASTER_OPEN_REGION,
          conf.getInt("hbase.master.executor.openregion.threads", 5));
       this.service.startExecutorService(ExecutorType.MASTER_CLOSE_REGION,
          conf.getInt("hbase.master.executor.closeregion.threads", 5));
       this.service.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
          conf.getInt("hbase.master.executor.serverops.threads", 5));
       this.service.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
          conf.getInt("hbase.master.executor.serverops.threads", 5));
       this.service.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS,
          conf.getInt("hbase.master.executor.logreplayops.threads", 10));
     
       // We depend on there being only one instance of this executor running
       // at a time.  To do concurrency, would need fencing of enable/disable of
       // tables.
       // Any time changing this maxThreads to > 1, pls see the comment at
       // AccessController#postCreateTableHandler
       this.service.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
       startProcedureExecutor();
     
        // Initial cleaner chore
        CleanerChore.initChorePool(conf);
       // Start log cleaner thread
        //获取定时日志清理时间,从系统配置获取,默认为10分钟
       int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 60 * 1000);
       this.logCleaner =
          new LogCleaner(cleanerInterval,
             this, conf, getMasterFileSystem().getOldLogDir().getFileSystem(conf),
             getMasterFileSystem().getOldLogDir());
       //将任务加入定时执行,时间间隔为 cleanerInterval ,该值在LogCleaner中已经设置为定时执行间隔
        getChoreService().scheduleChore(logCleaner);
     
       //start the hfile archive cleaner thread
        Path archiveDir = HFileArchiveUtil.getArchivePath(conf);
        Map<String, Object> params = new HashMap<String, Object>();
        params.put(MASTER, this);
        this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem()
            .getFileSystem(), archiveDir, params);
        getChoreService().scheduleChore(hfileCleaner);
        serviceStarted = true;
        if (LOG.isTraceEnabled()) {
          LOG.trace("Started service threads");
        }
        if (!conf.getBoolean(HConstants.ZOOKEEPER_USEMULTI, true)) {
          try {
            replicationZKLockCleanerChore = new ReplicationZKLockCleanerChore(this, this,
                cleanerInterval, this.getZooKeeper(), this.conf);
            getChoreService().scheduleChore(replicationZKLockCleanerChore);
          } catch (Exception e) {
            LOG.error("start replicationZKLockCleanerChore failed", e);
          }
        }
        try {
          replicationZKNodeCleanerChore = new ReplicationZKNodeCleanerChore(this, cleanerInterval,
              new ReplicationZKNodeCleaner(this.conf, this.getZooKeeper(), this));
          getChoreService().scheduleChore(replicationZKNodeCleanerChore);
        } catch (Exception e) {
          LOG.error("start replicationZKNodeCleanerChore failed", e);
        }
      }
    

    定时执行

    其中这段代码是对我们HLog进行处理,并加入调度定时执行

     // Initial cleaner chore
        CleanerChore.initChorePool(conf);
       // Start log cleaner thread
        //获取定时日志清理时间,从系统配置获取,默认为10分钟
       int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 60 * 1000);
       this.logCleaner =
          new LogCleaner(cleanerInterval,
             this, conf, getMasterFileSystem().getOldLogDir().getFileSystem(conf),
             getMasterFileSystem().getOldLogDir());
       //将任务加入定时执行,时间间隔为 cleanerInterval ,该值在LogCleaner中已经设置为定时执行间隔
        getChoreService().scheduleChore(logCleaner);
    

    加入调度后会周期性执行 LogCleaner.chore() 方法(在父类CleanerChore中)

     @Override
      protected void chore() {
        if (getEnabled()) {
          try {
            POOL.latchCountUp();
            if (runCleaner()) {
              if (LOG.isTraceEnabled()) {
                LOG.trace("Cleaned all WALs under " + oldFileDir);
              }
            } else {
              if (LOG.isTraceEnabled()) {
                LOG.trace("WALs outstanding under " + oldFileDir);
              }
            }
          } finally {
            POOL.latchCountDown();
          }
          // After each cleaner chore, checks if received reconfigure notification while cleaning.
          // First in cleaner turns off notification, to avoid another cleaner updating pool again.
          if (POOL.reconfigNotification.compareAndSet(true, false)) {
            // This cleaner is waiting for other cleaners finishing their jobs.
            // To avoid missing next chore, only wait 0.8 * period, then shutdown.
            POOL.updatePool((long) (0.8 * getTimeUnit().toMillis(getPeriod())));
          }
        } else {
          LOG.trace("Cleaner chore disabled! Not cleaning.");
        }
      }
    

    上面代码中的runCleaner()方法就是将我们CleanerTask加入任务队列中

      public Boolean runCleaner() {
        CleanerTask task = new CleanerTask(this.oldFileDir, true);
        POOL.submit(task);
        return task.join();
      }
    

    LogCleaner 日志清理类

    LogCleaner类是清理日志数据,LogCleaner 父类 CleanerChore 类中的 私有类CleanerTask(该类继承RecursiveTask类,不做过多介绍,想了解的可以百度 ForkJoinTask ), 的 compute()方法是定时清理的关键,这里获取了所有oldWALs目录下的文件,并进行选择性删除

    @Override
        protected Boolean compute() {
          LOG.trace("Cleaning under " + dir);
          List<FileStatus> subDirs;
          List<FileStatus> tmpFiles;
          final List<FileStatus> files;
          try {
            // if dir doesn't exist, we'll get null back for both of these
            // which will fall through to succeeding.
            subDirs = FSUtils.listStatusWithStatusFilter(fs, dir, new FileStatusFilter() {
              @Override
              public boolean accept(FileStatus f) {
                return f.isDirectory();
              }
            });
            if (subDirs == null) {
              subDirs = Collections.emptyList();
            }
            //获取oldWALs目录下文件
            tmpFiles = FSUtils.listStatusWithStatusFilter(fs, dir, new FileStatusFilter() {
              @Override
              public boolean accept(FileStatus f) {
                return f.isFile();
              }
            });
            files = tmpFiles == null ? Collections.<FileStatus>emptyList() : tmpFiles;
          } catch (IOException ioe) {
            LOG.warn("failed to get FileStatus for contents of '" + dir + "'", ioe);
            return false;
          }
     
          boolean allFilesDeleted = true;
          if (!files.isEmpty()) {
            allFilesDeleted = deleteAction(new Action<Boolean>() {
              @Override
              public Boolean act() throws IOException {
                //files 是oldWALs目录下所有文件
                return checkAndDeleteFiles(files);
              }
            }, "files");
          }
     
          boolean allSubdirsDeleted = true;
          if (!subDirs.isEmpty()) {
            final List<CleanerTask> tasks = Lists.newArrayListWithCapacity(subDirs.size());
            for (FileStatus subdir : subDirs) {
              CleanerTask task = new CleanerTask(subdir, false);
              tasks.add(task);
              //任务
              task.fork();
            }
            allSubdirsDeleted = deleteAction(new Action<Boolean>() {
              @Override
              public Boolean act() throws IOException {
                return getCleanResult(tasks);
              }
            }, "subdirs");
          }
     
          boolean result = allFilesDeleted && allSubdirsDeleted;
          // if and only if files and subdirs under current dir are deleted successfully, and
          // it is not the root dir, then task will try to delete it.
          if (result && !root) {
            result &= deleteAction(new Action<Boolean>() {
              @Override
              public Boolean act() throws IOException {
                return fs.delete(dir, false);
              }
            }, "dir");
          }
          return result;
        }
    

    上一步中调用了 checkAndDeleteFiles(files) 方法,该方法的作用是:通过每个清理程序运行给定的文件,以查看是否应删除该文件,并在必要时将其删除。输入参数是所有oldWALs目录下的文件

     /**
       * Run the given files through each of the cleaners to see if it should be deleted, deleting it if
       * necessary.
       * 通过每个清理程序运行给定的文件,以查看是否应删除该文件,并在必要时将其删除。
       * @param files List of FileStatus for the files to check (and possibly delete)
       * @return true iff successfully deleted all files
       */
      private boolean checkAndDeleteFiles(List<FileStatus> files) {
        if (files == null) {
          return true;
        }
     
        // first check to see if the path is valid
        List<FileStatus> validFiles = Lists.newArrayListWithCapacity(files.size());
        List<FileStatus> invalidFiles = Lists.newArrayList();
        for (FileStatus file : files) {
          if (validate(file.getPath())) {
            validFiles.add(file);
          } else {
            LOG.warn("Found a wrongly formatted file: " + file.getPath() + " - will delete it.");
            invalidFiles.add(file);
          }
        }
     
        Iterable<FileStatus> deletableValidFiles = validFiles;
        // check each of the cleaners for the valid files
        for (T cleaner : cleanersChain) {
          if (cleaner.isStopped() || getStopper().isStopped()) {
            LOG.warn("A file cleaner" + this.getName() + " is stopped, won't delete any more files in:"
                + this.oldFileDir);
            return false;
          }
     
          Iterable<FileStatus> filteredFiles = cleaner.getDeletableFiles(deletableValidFiles);
          
          // trace which cleaner is holding on to each file
          if (LOG.isTraceEnabled()) {
            ImmutableSet<FileStatus> filteredFileSet = ImmutableSet.copyOf(filteredFiles);
            for (FileStatus file : deletableValidFiles) {
              if (!filteredFileSet.contains(file)) {
                LOG.trace(file.getPath() + " is not deletable according to:" + cleaner);
              }
            }
          }
          
          deletableValidFiles = filteredFiles;
        }
        
        Iterable<FileStatus> filesToDelete = Iterables.concat(invalidFiles, deletableValidFiles);
        return deleteFiles(filesToDelete) == files.size();
      }
    

    ReplicationLogCleaner 日志清理类

    checkAndDeleteFiles方法中 又调用了 cleaner.getDeletableFiles(deletableValidFiles) ,getDeletableFiles方法在ReplicationLogCleaner类下,是判断哪些文件该删除,哪些不该删除,删除条件就是文章开头提出的是否在参与复制中,如果在参与则不删除,不在则删除。
    注:所有在参与peer的数据都在 zookeeper 中 /hbase/replication/rs 目录下存储

    比如在zookeeper目录下有这么个节点

    /hbase/replication/rs/jast.zh,16020,1576397142865/Indexer_account_indexer_prd/jast.zh%2C16020%2C1576397142865.jast.zh%2C16020%2C1576397142865.regiongroup-0.1579283025645
    

    那么我们再oldWALs目录下是不会删除掉这个数据的

    [jast@jast002 ~]$ hdfs dfs -du -h /hbase/oldWALs/jast015.zh%2C16020%2C1576397142865.jast015.zh%2C16020%2C1576397142865.regiongroup-0.1579283025645
    256.0 M  512.0 M  /hbase/oldWALs/jast015.zh%2C16020%2C1576397142865.jast015.zh%2C16020%2C1576397142865.regiongroup-0.1579283025645
    
     @Override
      public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
       // all members of this class are null if replication is disabled,
       // so we cannot filter the files
        if (this.getConf() == null) {
          return files;
        }
     
        final Set<String> wals;
        try {
          // The concurrently created new WALs may not be included in the return list,
          // but they won't be deleted because they're not in the checking set.
          wals = loadWALsFromQueues();
        } catch (KeeperException e) {
          LOG.warn("Failed to read zookeeper, skipping checking deletable files");
          return Collections.emptyList();
        }
     
        return Iterables.filter(files, new Predicate<FileStatus>() {
          @Override
          public boolean apply(FileStatus file) {
            String wal = file.getPath().getName();
            //包含文件则保留,不包含则删除
            boolean logInReplicationQueue = wals.contains(wal);
            if (LOG.isDebugEnabled()) {
              if (logInReplicationQueue) {
                //包含文件保留
                LOG.debug("Found log in ZK, keeping: " + wal);
              } else {
                //不包含删除
                LOG.debug("Didn't find this log in ZK, deleting: " + wal);
              }
            }
           return !logInReplicationQueue;
          }});
      }
    

    上一步调用了 loadWALsFromQueues 方法,该方法作用是:获取所有在复制队列中的wals文件,并返回,

    /**
       * Load all wals in all replication queues from ZK. This method guarantees to return a
       * snapshot which contains all WALs in the zookeeper at the start of this call even there
       * is concurrent queue failover. However, some newly created WALs during the call may
       * not be included.
       *
       * 从ZK加载所有复制队列中的所有wals。 即使存在并发队列故障转移,
       * 此方法也保证在此调用开始时返回包含zookeeper中所有WAL的快照。
       * 但是,可能不会包括通话过程中一些新创建的WAL。
       */
      private Set<String> loadWALsFromQueues() throws KeeperException {
        for (int retry = 0; ; retry++) {
          int v0 = replicationQueues.getQueuesZNodeCversion();
          List<String> rss = replicationQueues.getListOfReplicators();
          if (rss == null || rss.isEmpty()) {
            LOG.debug("Didn't find any region server that replicates, won't prevent any deletions.");
            return ImmutableSet.of();
          }
          Set<String> wals = Sets.newHashSet();
          for (String rs : rss) {
            //加载zookeeper下,/hbase/replication/rs 目录下所有数据
            List<String> listOfPeers = replicationQueues.getAllQueues(rs);
            // if rs just died, this will be null
            if (listOfPeers == null) {
              continue;
            }
            //加载所有目录
            for (String id : listOfPeers) {
              List<String> peersWals = replicationQueues.getLogsInQueue(rs, id);
              if (peersWals != null) {
                wals.addAll(peersWals);
              }
            }
          }
          int v1 = replicationQueues.getQueuesZNodeCversion();
          if (v0 == v1) {
            return wals;
          }
          LOG.info(String.format("Replication queue node cversion changed from %d to %d, retry = %d",
              v0, v1, retry));
        }
      }
    

    总结

    至此我们可以发现,删除的过程就是定期执行删除文件线程,从oldWALs获取所有文件,如果在peer复制队列中则不进行副本删除,否则则删除

    展开全文
  • HBase 预写日志 (WAL)

    2011-11-10 15:07:08
    HBase数据格式包括Log结构参考上一篇日志。 WAL最重要的作用是灾难恢复,一旦服务器崩溃,通过重放log,我们可以恢复崩溃之前的数据。如果写入WAL失败,整个操作也将认为失败。 图6 WAL   基本流程:首先...

     

    HBase数据格式包括Log结构参考上一篇日志。

    WAL最重要的作用是灾难恢复,一旦服务器崩溃,通过重放log,我们可以恢复崩溃之前的数据。如果写入WAL失败,整个操作也将认为失败。

    6  WAL

     

    基本流程:首先,客户端初始化一个可能对数据改动的操作,如put(Put),delete(Delete)  incrementColumnValue()。这些操作将被封装在一个KeyValue对象实例中,发送给HRegionServer 一旦达到一定大小,HRegionServer 将其发送给HRegion。这个过程中,数据会首先会被写入WAL,之后将被写到实际存放数据的MemStore中。

     

    HLog是实现WAL的类。一个HRegionServer对应一个HLog实例。当HRegion初始化时,HLog将作为一个参数传给HRegion的构造函数。

    WAL中几个重要的类:

    1.      HLog

    HLog最核心的是调用doWriteappend() 方法,任何对数据改动的操作都就将首先调用这个方法。

    出于性能考虑,put(), delete() incrementColumnValue()可通过开关函数setWriteToWAL (boolean) 禁用WAL。运行MapReduce Job时,可通过关闭WAL获得性能提升。

    HLog另一个重要特性是将通过sequence number追踪数据改变。它内部使用AtomicLong保证线程安全。

    2.      HLogKey

    上一部分的存储格式中有提到,HLogKey包含的一些主要变量,主要用于记录Key/Value对的归属信息。

    3.      LogFlusher

    数据以KeyValue形式到达HRegionServer,将写入WAL,之后写入一个SequenceFile。由于数据流在写入文件时经常会缓存以提高性能,因此,有时数据实际保存在内存中。

    LogFlusher调用HLog.optionalSync(),后者根据hbase. regionserver. optionallogflushinterval (默认10)定期调用Hlog.sync()HLog.doWrite()也根据hbase.regionserver.flushlogentries  (默认100)定期调用Hlog.sync()Sync() 本身调用HLog.Writer.sync(),它由SequenceFileLogWriter实现。

    4.      LogRoller

    Log的大小通过$HBASE_HOME/conf/hbase-site.xml hbase.regionserver.logroll.period限制,默认是一个小时。所以每60分钟,会打开一个新的log文件。久而久之,会有一大堆的文件需要维护。LogRoller主要完成日志的清理。

    首先,LogRoller调用HLog.rollWriter(),定时滚动日志,之后,利用HLog.cleanOldLogs()可以清除旧的日志。它首先取得存储文件中的最大的sequence number,之后检查是否存在一个log所有的条目的”sequence number”均低于这个值,如果存在,将删除这个log

    5.      Replay

    旧日志往往由RegionServer 崩溃产生。当HMaster启动或者检测到RegionServer 崩溃,它将日志文件拆分为多份文件,存储在region所属的文件夹。之后,将日志重放。

    重放过程:HRegionServer启动,打开所管辖的Region,检查是否存在剩余的log文件,如果存在,调用Store.doReconstructionLog()。重放一个日志只是简单地读入一个日志,将日志中的条目加入到Memstore中。最后,flush操作将Memstore中数据flush到硬盘中。


     

    展开全文
  • HBase WAL预写日志

    千次阅读 2016-12-12 09:48:08
    HBase中使用预写日志(WAL)的方式来解决这一问题:每次更新都会写入日志,只有写入成功才会通知客户端操作成功,然后服务器可以按需自由地批量处理或聚合内存中的数据。 下面图展示WSL是怎样和HB

    HBase的region服务器会将数据保存到内存中,直到有足够多的数据才会将其刷写到硬盘中,这样可以避免创建很多的小文件。存储在内存中的数据具有不稳定性因素,因此如果服务器断电了那么内存中的数据就会丢失。
    HBase中使用预写日志(WAL)的方式来解决这一问题:每次更新都会写入日志,只有写入成功才会通知客户端操作成功,然后服务器可以按需自由地批量处理或聚合内存中的数据。
    下图展示WAL是怎样和HBase的架构结合在一起的,

    这里写图片描述

    从上图可以看出,WAL被同一个region服务器的所有region共享,所以对于每一次修改它就像一个日志中心一样。
    整个处理过程如下:
    1 客户端启动一个操作来修改数据,如put(),delet()等。
    2 每一个修改被封装到一个KeyValue对象实例,并通过RPC调用发送出来。
    3 上述调用成批地发送给含有匹配region的HRegionServer。
    4 数据先被写入到WAL,然后被放放到实际拥有记录的存储文件的MemStore中。
    5 当MemStore达到一定的大小或经历一个特定时间之后,数据会异步地连续写入到文件系统中。

    由于实际的日志存储在HDFS上,所以即使在服务器完全崩溃的情况下,WAL也能保证数据不会丢失。其他服务器可以打开日志文件然后回放这些修改。

    展开全文
  • HBase 架构101 –预写日志系统 (WAL)

    千次阅读 2011-11-09 18:11:26
    HBase 架构101 –预写日志系统 (WAL) 原文:http://www.larsgeorge.com/2010/01/hbase-architecture-101-write-ahead-log.html 什么是预写日志WAL? 之前的文章我们简单介绍了HBase的存储结构。其中提到了预写...
  • hbase记录日志walWrite-Ahead Logging has been the default journaling mode for Core Data SQLite stores since iOS 7 and OS X Mavericks. Journaling in Core Data is best explained as the way data ...
  • HBase 架构101 –预写日志系统 (WAL)什么是预写日志WAL? 之前的文章我们简单介绍了HBase的存储结构。其中提到了预写日志。这里,我们要介绍它的实现细节,所有的描述都基于HBase 0.20.3.WAL最重要的作用是灾难恢复...
  • hbase记录日志wal SQL Server transaction log is one of the most critical and in the same time one of the most misinterpreted part. While being neglected, it can easily become a bottleneck to our SQL ...
  • WAL预写日志WAL的目的是为了容灾。一个Region Server会将本节点上所有的region的修改记录都记到同一个HLog文件中。当一次操作成功记录到WAL中时才会返回成功。这种机制保证了断电不会丢失数据。当region打开时,...
  • HBase为什么先写内存再写日志 HBase提供了一个MVCC机制,来保障写数据阶段的数据可见性。数据先写入Memstore再刷新WAL,是为了一些特殊场景下,内存中的数据能够更及时可见。如果写入WAL失败的话,Memstore中的数据...
  • Hbase日志系统

    2020-09-04 10:52:35
    本文介绍了HBase中的日志系统,搞清楚了HBase高可用性、备份、背后的实现原理,以及HBase是如何刷写日志的,通过哪几个存储模块的配合来完成日志的刷写的。
  • hbase架构及hbase的读写过程 待会总结 先看学习网址 学习网址:https://blog.csdn.net/whdxjbw/article/details/81107285
  • 预写日志(WAL,Write-Ahead Log)将每次状态更新抽象为一个命令并追加写入一个日志中,这个日志只追加写入,也就是顺序写入,所以 IO 会很快。相比于更新存储的数据结构并且更新落盘这个随机 IO 操作,写入速度更快...
  • HBase当中的数据最终都是存储在HDFS上面的,HBase天生的支持MR的操作,我们可以通过MR直接处理HBase当中的数据,并且MR可以将处理后的结果直接存储到HBase当中去 需求:读取HBase当中一张表的数据,然后将数据写入到...
  • HBase的读写流程

    2019-03-27 20:00:10
    HBase的读写流程 我们看到HBase集群的物理模型,包括:Client、ZooKeeper、HMaster、HRegionServer、HLog、HRegion、Store、StoreFile、MemStore。对于HBase,它的元数据存放在ZooKeeper中,真实数据存放在...
  • 文章目录15、HBase分区1、为何要分区?2、如何分区?3、如何设定分区?1、手动指定分区2、使用16进制算法生成分区3、使用JavaAPI创建...
  • HBase学习日志

    2020-09-10 18:37:04
    记一下学习HBase时学了什么。 HBase定义,一种数据库。支持增删改查操作。 HBase逻辑结构:列、列族(多个列)、row key(唯一标识)、region(多个行,即表的切片内容)、store(真正存储的内容)。下图可以视作三...
  • Hbase读写

    2019-11-12 19:41:35
    负责管理HBase元数据,即表的结构、表存储的Region等元信息。 负责表的创建,删除和修改(因为这些操作会导致HBase元数据的变动)。 负责为HRegionServer分配Region,分配好后也会将元数据写入相应位置(后面会详细...
  • Hbase写流程

    2021-04-17 19:01:56
    在向Hbase中写数据的时候,我们操作Client进行,...数据会先写到预写日志WAL中,之后regionserver会立即返回写入成功的消息,而数据在regionserver中的WAL保留,由memstore对数据做字典排序,最后写入storefile ...
  • HBase数据流程

    2020-05-26 11:18:05
    1.客户端访问zookeeper获取hbase:meta表所在...3.与目标RegionServer通信,先将数据追加到写入日志(WAL)中,再写入对应的缓存中,数据在缓存中进行排序。(若写入日志同步到hdfs成功,缓存中的数据有效,数
  • HbaseHbaseHbase读写流程1、Hbase读流程2、Hbase写流程RowKey设计原则1、散列原则2、唯一原则3、长度原则Hbase二级索引重置Hbase Hbase bigTable NoSQL数据库 列式数据库 实时查询的数据库 Hbase - Hadoop...
  • hbase的读写流程

    2019-09-22 22:10:46
    hbase的读写流程 1)Client先访问zookeeper,从meta表读取region的位置,然后读取meta表中的数据。meta中又存储了用户表的region信息; 2)根据namespace、表名和rowkey在meta表中找到对应的region信息; 3)找到这...
  • 然后访问hmasterhmaster会将处理这次操作的HRegionServer服务的地址给client,然后client会去请求对应的HRegionServer如果设置了AWL(Write-Ahead-Log)预写日志,那么HRegionServer会先进行预写日志操作,并

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 7,874
精华内容 3,149
关键字:

hbase预写日志