HBase优化、协处理器、容灾

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * <p>
 *
 * </p>
 *
 * @author guazi
 * Email guazi@uoko.com
 * created by 2019/07/22
 */
public class RegionServerTest extends BaseRegionObserver {

    private byte[] columnFamily = Bytes.toBytes("cf");

    private byte[] countCol = Bytes.toBytes("countCol");

    private byte[] unDeleteCol = Bytes.toBytes("unDeleteCol");

    private RegionCoprocessorEnvironment environment;

    // RegionServer 打开region时执行
    @Override
    public void start(CoprocessorEnvironment e) throws IOException {
        super.start(e);
    }
    // RegionServer 关闭region时执行
    @Override
    public void stop(CoprocessorEnvironment e) throws IOException {
        super.stop(e);
    }

    /**
     * 1. cf : countCol 进行累加操作 每次插入时,都要与之前的值进行相加
     */
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {

        if (put.has(columnFamily,countCol)){
            // 如果发现新插入的包含countCol这一列
            // 获取old countCol value
            Result rs = e.getEnvironment().getRegion().get(
                    new Get(put.getRow())
            );

            int oldNum = 0;

            for (Cell cell:rs.rawCells()){
                if (CellUtil.matchingColumn(cell,columnFamily,countCol)){
                    oldNum = Integer.valueOf(Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }

            // 获取new countCol value
            List<Cell> cells = put.get(columnFamily,countCol);
            int newNum = 0;
            for (Cell cell : cells){
                if (CellUtil.matchingColumn(cell,columnFamily,countCol)){
                    newNum = Integer.valueOf(Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }

            // sum AND update
            put.addColumn(columnFamily,countCol,Bytes.toBytes(
                    String.valueOf(oldNum + newNum)
            ));
        }
    }

    /**
     * 2. 不能直接删除unDeleteCol 删除countCol的时候将unDeleteCol一起删除
     */
    @Override
    public void preDelete(ObserverContext<RegionCoprocessorEnvironment> e,
                          Delete delete, WALEdit edit, Durability durability) throws IOException {
        // 判断是否操作cf列
        List<Cell> cells = delete.getFamilyCellMap().get(columnFamily);

        if (CollectionUtils.isEmpty(cells))return;

        boolean deleteFlag = false;

        for (Cell cell : cells) {
            byte[] qualifier = CellUtil.cloneQualifier(cell);

            if (Arrays.equals(qualifier, unDeleteCol)) {
                throw new IOException("can not delete unDel column");
            }
            if (Arrays.equals(qualifier, countCol)) {
                deleteFlag = true;
            }
        }

        if (deleteFlag){
            delete.addColumn(columnFamily,unDeleteCol);
        }
    }
}

public class TestRowCountEndPoint extends hbaseEndPointTestService
                implements Coprocessor, CoprocessorService {

                @Override
                ...
}

1. 创建新的表并指定相应列族
2. 执行命令进行拷贝

./hbase org.apache.hadoop.hbase.mapreduce.CopyTable --new.name=FileTableNew FileTable

Export

./hbase org.apache.hadoop.hbase.mapreduce.Export {导出的表名} {存储在hdfs的路径}

Import

./hbase org.apache.hadoop.hbase.mapreduce.Import {导入的表名} {存储在hdfs的路径}

<property>
    <name>hbase.snapshot.enabled</name>
    <value>true</value>
</property>

#创建快照
snapshot 'myTable','myTableSnapshot-190731'
#克隆快照
clone_snapshot 'myTableSnapshot-190731','myNewTestTable'
#列出快照
list_snapshots
#删除快照
delete_snapshot 'myTableSnapshot-190731'
#恢复数据
disable 'myTable'
restore_snapshot 'myTableSnapshot-190731'

hbase-site.xml

<property>
    <name>hbase.replication</name>
    <value>true</value>
</property>

#在原集群及目标集群都创建同名表
#指定目标集群zk地址和路径

# 执行命令
add_peer '1',"zk01:2181:/hbase_backup"

#标注需要备份的列族信息及备份的目标库地址
# replication_scope值为上面add_peer指定的值

alter 'replication_source_table',{NAME=>'f1',REPLICATION_SCOPE=>'1'}