spring-data-hbase的使用

 

maven的配置

<!-- hbase start -->
<dependency>
    <groupId>org.springframework.data</groupId>
    <artifactId>spring-data-hadoop</artifactId>
    <version>${spring.hadoop.version}</version>
    <exclusions>
        <exclusion>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context-support</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>javax.servlet</groupId>
            <artifactId>servlet-api</artifactId>
        </exclusion>
        <exclusion>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
        </exclusion>
    </exclusions>
</dependency>


<!-- hadoop start -->
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-client</artifactId>
    <version>${hadoop.version}</version>
    <scope>compile</scope>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<!-- hadoop end -->

<!-- hbase start -->
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>${hbase.version}</version>
    <exclusions>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>javax.servlet</groupId>
            <artifactId>servlet-api</artifactId>
        </exclusion>
        <!-- <exclusion>
            <groupId>javax.servlet.jsp</groupId>
            <artifactId>jsp-api</artifactId>
        </exclusion> -->
        <exclusion>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<!-- hbase end -->

 

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.hadoop.hbase.HbaseTemplate;

import java.io.IOException;

/**
 * HbaseTempalteConfig
 *
 * @ClassName: HbaseTempalteConfig
 * @author kanpiaoxue
 * @version 1.0
 * @CreateTime: 2020/04/10 14:19:41
 * @Description:
 */
@Configuration
public class HbaseTempalteConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(HbaseTempalteConfig.class);

    @Value("${hbase.zookeeper.quorum}")
    private String zkQuorum;
    @Value("${hbase.zookeeper.port:2181}")
    private Integer zkPort;
    @Value("${hbase.zookeeper.znode.parent}")
    private String zkZnodeParent;

    @Bean
    public Admin hbaseAdmin(@Autowired org.apache.hadoop.conf.Configuration configuration)
            throws IOException {
        LOGGER.debug("start to hbaseAdmin. zkQuorum:{},zkPort:{},zkZnodeParent:{}", zkQuorum, zkPort,
                zkZnodeParent);
        Connection connection = ConnectionFactory.createConnection(configuration);
        Admin admin = connection.getAdmin();
        return admin;
    }

    @Bean
    public org.apache.hadoop.conf.Configuration hbaseConfiguration() {
        LOGGER.debug("start to hbaseConfiguration. zkQuorum:{},zkPort:{},zkZnodeParent:{}", zkQuorum, zkPort,
                zkZnodeParent);
        org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
        configuration.set(HConstants.ZOOKEEPER_QUORUM, zkQuorum);
        configuration.set(HConstants.ZOOKEEPER_CLIENT_PORT, String.valueOf(zkPort));
        configuration.set(HConstants.ZOOKEEPER_ZNODE_PARENT, zkZnodeParent);
        configuration.set("log4j.logger.org.apache.hadoop.hbase", "DEBUG");
        // 设置keyvalue.maxsize的大小限制，这里设置500M = 500*1024*1024
        configuration.set("hbase.client.keyvalue.maxsize", "524288000");

        return configuration;
    }

    @Bean
    public HbaseTemplate hbaseTemplate(@Autowired org.apache.hadoop.conf.Configuration configuration) {
        LOGGER.debug("start to hbaseTemplate. zkQuorum:{},zkPort:{},zkZnodeParent:{}", zkQuorum, zkPort,
                zkZnodeParent);
        HbaseTemplate hbaseTemplate = new HbaseTemplate(configuration);

        return hbaseTemplate;
    }

}

 

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Service;

import com.google.common.base.Preconditions;
import com.google.common.io.ByteSource;
import com.google.common.io.Files;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Objects;

import javax.annotation.PostConstruct;

/**
 * HBase的文件服务实现类
 *
 * @ClassName: HBaseRemoteTargetFileServiceImpl
 * @author kanpiaoxue
 * @version 1.0
 * @CreateTime: 2020/04/10 16:12:51
 * @Description:
 */
@Service
public class HBaseRemoteTargetFileServiceImpl implements RemoteTargetFileService {
    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseRemoteTargetFileServiceImpl.class);

    @Autowired
    private Admin admin;
    @Autowired
    private HbaseTemplate hbaseTemplate;

    @Value("${hbase.table.name}")
    private String tableName;
    @Value("${hbase.table.column.family}")
    private String columnFamily;
    @Value("${hbase.table.column.columnName}")
    private String column;

    private byte[] tableNameAsBytes;
    private byte[] columnFamilyAsBytes;
    private byte[] columnAsBytes;

 
    @Override
    public InputStream getFile(String key) throws Exception {
        LOGGER.info("start to getFile. key:{}", key);
        key = CommonUtils.checkStringArgs(key, "key");
        InputStream input = hbaseTemplate.get(tableName, key, (Result result, int rowNum) -> {
            byte[] value = result.getValue(columnFamilyAsBytes, columnAsBytes);
            if (Objects.isNull(value)) {
                return null;
            }
            InputStream targetStream = ByteSource.wrap(value).openStream();
            return targetStream;
        });

        CommonUtils.checkDataExist(input,
                String.format(
                        "can not find data with tableName:%s,columnFamily:%s,column:%s,key:%s from hbase",
                        tableName, columnFamily, column, key));

        return input;
    }

    @PostConstruct
    public void initialize() throws Exception {
        LOGGER.info("start to initialize. tableName:{},columnFamily:{},column:{},admin:{},hbaseTemplate:{}",
                tableName, columnFamily, column, admin, hbaseTemplate);
        Preconditions.checkNotNull(admin, "admin is null");
        Preconditions.checkNotNull(hbaseTemplate, "hbaseTemplate is null");
        initTableInfo();
        initTable();
        LOGGER.info("finish initializing. tableName:{},columnFamily:{},column:{},admin:{},hbaseTemplate:{}",
                tableName, columnFamily, column, admin, hbaseTemplate);
    }


    @Override
    public void saveFile(String key, File file) throws Exception {
        LOGGER.info("start to saveFile. key:{},file:{}", key, file);
        CommonUtils.checkStringArgs(key, "key");
        Preconditions.checkNotNull(file, "file is null");
        Preconditions.checkArgument(file.exists(), "file:%s does not exist.", file);
        Preconditions.checkArgument(file.isFile(), "file:%s is not a valid file.", file);
        try (InputStream input = Files.asByteSource(file).openStream()) {
            byte[] value = IOUtils.toByteArray(input);
            hbaseTemplate.put(tableName, key, columnFamily, column, value);
        }
    }

    private void initTable() throws IOException {
        LOGGER.info("start to initializeTable.");
        TableName tableName = TableName.valueOf(tableNameAsBytes);
        if (!admin.tableExists(tableName)) {
            LOGGER.info("tableName:{} does not exist. create it.", tableName);
            HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
            tableDescriptor.addFamily(columnDescriptor);
            admin.createTable(tableDescriptor);
        }
    }

    private void initTableInfo() {
        LOGGER.info("start to initTableInfo.");
        tableName = CommonUtils.checkStringArgs(tableName, "tableName");
        columnFamily = CommonUtils.checkStringArgs(columnFamily, "columnFamily");
        column = CommonUtils.checkStringArgs(column, "column");

        tableNameAsBytes = Bytes.toBytes(tableName);
        columnFamilyAsBytes = Bytes.toBytes(columnFamily);
        columnAsBytes = Bytes.toBytes(column);
    }

}

 

参考文档：

1、https://www.baeldung.com/hbase

2、https://www.jianshu.com/p/c04993cf85ac

3、https://docs.spring.io/spring-hadoop/docs/1.0.0.RELEASE/reference/html/hbase.html