Bulkload Hive表到HBase

水深无声 2022-07-26 00:06 226阅读 0赞

# 1.描述 #

HBase可以随机读写海量的数据，但是如果把这海量数据导入到HBase却是一个挑战。如，将Hive表尽可能快的导入到HBase中。这里有以下三种解决方案：

1.  使用API把数据一条一条地写入HBase。
2.  用[***HBaseIntegration***][HBaseIntegration]方法。
3.  使用HBase自带的Bulkload功能。

但是，第一种方法明显是最低效的；第二种方法我之前已经提到过，同样比较慢；那么，第三种方法呢？毋庸置疑，Bulkload方法是最高效的解决方案。因为Bulkload方法可以让你绕过创建HFile（HBase Data Files），而是直接把已经生成的HFile拷贝到HDFS上。如果你的Hive和HBase部署在同一个集群上时，Bulkload的处理过程包括两个步骤：

1.  用MapReduce生成HFile。
2.  使用LoadIncrementalHFiles.doBulkLoad将HFile导入到HBase。

public class Driver extends Configured implements Tool{
    
        private static Configuration conf = new Configuration();
        private static Configuration hconf = null;
        private static HBaseAdmin hadmin = null;
    
        public static void connectHBase(){
            final String HBASE_CONFIG_ZOOKEEPER_CLIENT = "hbase.zookeeper.property.clientPort";
            final String HBASE_ZOOKEEPER_CLIENT_PORT = "2181";
            final String HBASE_CONFIG_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
            final String HBASE_ZOOKEEPER_SERVER = "hbase38,hbase43,hbase00";
    
            conf.set(HBASE_CONFIG_ZOOKEEPER_CLIENT, HBASE_ZOOKEEPER_CLIENT_PORT);
            conf.set(HBASE_CONFIG_ZOOKEEPER_QUORUM, HBASE_ZOOKEEPER_SERVER);
            hconf = HBaseConfiguration.create(conf);
            try{
                hadmin = new HBaseAdmin(hconf);
            }
            catch (Exception e){
                e.printStackTrace();
            }
        }
    
    
        public static void main(String[] args)throws Exception{
            String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if(otherArgs.length != 4){
                System.err.println("Usage: <rcfile> <hfile> <schemafile> <hbasetable>");
                System.exit(1);
            }
    
            String path = System.getProperty("user.dir") + otherArgs[2];
            List<String> fieldNames = HiveTableUtils.getFieldName(path);
            StringBuilder sb = new StringBuilder(fieldNames.get(0));
            int size = fieldNames.size();
            for(int i = 1; i < size; i++){
                sb.append(":").append(fieldNames.get(i));
            }
    
            conf.set("schema", sb.toString());
    		
    	if(ToolRunner.run(conf, new Driver(), otherArgs) == 0){
    		// Importing the generated HFiles into a HBase table
    		LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
    		loader.doBulkLoad(new Path(otherArgs[1], otherArgs[3]);
    		System.exit(0);
    	}
    	else{
    		System.exit(1);
    	}
        }
    
        @SuppressWarnings("deprecation")
        @Override
        public int run(String[] strings) throws Exception {
    
            Configuration config = getConf();
            Driver.connectHBase();
    
            Job job = new Job(config, "RCFile to HFile");
            job.setJarByClass(Driver.class);
            job.setMapperClass(RCFileToHFile.ParseMapper.class);
            job.setMapOutputKeyClass(ImmutableBytesWritable.class);
            job.setMapOutputValueClass(KeyValue.class);
    
            //Reduce's number is 0.
            job.setNumReduceTasks(0);
    
            job.setPartitionerClass(SimpleTotalOrderPartitioner.class);
    
            job.setInputFormatClass(RCFileMapReduceInputFormat.class);
    //		job.setOutputFormatClass(HFileOutputFormat.class);
    
            HTable table = new HTable(config, strings[3]);
            HFileOutputFormat.configureIncrementalLoad(job, table);
    
            RCFileMapReduceInputFormat.addInputPath(job, new Path(strings[0]));
            FileOutputFormat.setOutputPath(job, new Path(strings[1]));
    
            return job.waitForCompletion(true) ? 0 : 1;
        }
    }

然而，如果像我司那样——Hive和HBase部署在不同的集群上时，Bulkload的处理过程就需要三个步骤：

1.  用MapReduce生成HFile。  
    
2.  将已生成的HFile从Hive集群拷贝到HBase集群。
3.  用HBase的命令将HFile导入到HBase中（你也可以用Java代码如上述代码，这样就相对复杂了）。

我们先看第一步的代码：

public class Driver extends Configured implements Tool{
    
        private static Configuration conf = new Configuration();
        private static Configuration hconf = null;
        private static HBaseAdmin hadmin = null;
    
        public static void connectHBase(){
            final String HBASE_CONFIG_ZOOKEEPER_CLIENT = "hbase.zookeeper.property.clientPort";
            final String HBASE_ZOOKEEPER_CLIENT_PORT = "2181";
            final String HBASE_CONFIG_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
            final String HBASE_ZOOKEEPER_SERVER = "hbase38,hbase43,hbase00";
    
            conf.set(HBASE_CONFIG_ZOOKEEPER_CLIENT, HBASE_ZOOKEEPER_CLIENT_PORT);
            conf.set(HBASE_CONFIG_ZOOKEEPER_QUORUM, HBASE_ZOOKEEPER_SERVER);
            hconf = HBaseConfiguration.create(conf);
            try{
                hadmin = new HBaseAdmin(hconf);
            }
            catch (Exception e){
                e.printStackTrace();
            }
        }
    
    
        public static void main(String[] args)throws Exception{
            String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if(otherArgs.length != 4){
                System.err.println("Usage: <rcfile> <hfile> <schemafile> <hbasetable>");
                System.exit(1);
            }
    
            String path = System.getProperty("user.dir") + otherArgs[2];
            List<String> fieldNames = HiveTableUtils.getFieldName(path);
            StringBuilder sb = new StringBuilder(fieldNames.get(0));
            int size = fieldNames.size();
            for(int i = 1; i < size; i++){
                sb.append(":").append(fieldNames.get(i));
            }
    
            conf.set("schema", sb.toString());
    		
    		System.exit(ToolRunner.run(conf, new Driver(), otherArgs));
        }
    
        @SuppressWarnings("deprecation")
        @Override
        public int run(String[] strings) throws Exception {
    
            Configuration config = getConf();
            Driver.connectHBase();
    
            Job job = new Job(config, "RCFile to HFile");
            job.setJarByClass(Driver.class);
            job.setMapperClass(RCFileToHFile.ParseMapper.class);
            job.setMapOutputKeyClass(ImmutableBytesWritable.class);
            job.setMapOutputValueClass(KeyValue.class);
    
            //Reduce's number is 0.
            job.setNumReduceTasks(0);
    
            job.setPartitionerClass(SimpleTotalOrderPartitioner.class);
    
            job.setInputFormatClass(RCFileMapReduceInputFormat.class);
    //		job.setOutputFormatClass(HFileOutputFormat.class);
    
            HTable table = new HTable(config, strings[3]);
            HFileOutputFormat.configureIncrementalLoad(job, table);
    
            RCFileMapReduceInputFormat.addInputPath(job, new Path(strings[0]));
            FileOutputFormat.setOutputPath(job, new Path(strings[1]));
    
            return job.waitForCompletion(true) ? 0 : 1;
        }
    }

第二步，拷贝命令 [distcp][]：

# Distributed copy HFile to mycluster-hbase.
    hadoop distcp hdfs://mycluster-hive/hfile/hbase hdfs://mycluster-hbase/hbase/test

第三步，Bulkload：

# BulkLoad HFile into hbase table on mycluster-hbase.
    hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /hbase/test hbase_table

# 2.生成HFile #

MapReduce的作用是生成HFile。由于Hive采用的是RCFile的存储结构，所以我的MapReduce的输入数据是RCFile文件，你可以查看 [**MapReduce读/写RCFile文件**][MapReduce_RCFile]了解更多细节。值得一提的是，在生成HFile之前，你需要得到Hive表的schema，即字段名称。你可以通过以下方式得到Hive表的字段名。

*  解析存储Hive表元数据的文件。
 *  获取Hive表的元数据，
    
    1.  访问MySQL
    2.  用HCatalog访问Hive表

但是，我个人认为解析文件的方式更为高效，因为我们的Hive的metadata一般有数千列。

public class HiveTableUtils {
    
        //Gain hive table columns by parsing file.
        public static List<String>  getFieldName(String filePath){
            File file = new File(filePath);
            BufferedReader reader = null;
            List<String> fieldName = new ArrayList<String>();
    
            try {
                if (file.exists()) {
                    reader = new BufferedReader(new FileReader(file));
                    String tmp = null;
                    while ((tmp = reader.readLine()) != null) {
                        if (tmp.contains("`") && tmp.contains("COMMENT")) {
                            int start = tmp.indexOf("`");
                            int end = tmp.lastIndexOf("`");
                            fieldName.add(tmp.substring(start + 1, end));
                        }
                    }
                } else {
                    System.err.println("The file doesn't exist!");
                    System.exit(1);
                }
    
                reader.close();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
    
            return fieldName;
        }

至于Mapper的输入类型，就是 **ImmutableBytesWritable, KeyValue**。这些类使用了随后的Partitioner和Reducer来创建HFile。当然，你不需要实现自己的Reducer，代码中的 **HFileOutputFormat.configureIncrementalLoad()**会自动为你设置合理的Partitioner和Reducer。

public class RCFileToHFile {
    	
    	public static class ParseMapper extends Mapper<LongWritable, BytesRefArrayWritable, ImmutableBytesWritable, KeyValue>{
    //		private List<String> fieldName = null;
    		private String[] fieldName = null;
    
    		@Override
    		protected void setup(Context context)
    				throws IOException, InterruptedException {
    			// TODO Auto-generated method stub
    			super.setup(context);
    			Configuration conf = context.getConfiguration();
                
                String schema = conf.get("schema");
                fieldName = schema.split(":");
    
    //			fieldName = new ArrayList<String>();
    //			fieldName.add("id");
    //			fieldName.add("name");
    //			fieldName.add("age");
    		}
    		
    		@Override
    		protected void map(LongWritable key, BytesRefArrayWritable values,
    				Context context)
    				throws IOException, InterruptedException {
    			// TODO Auto-generated method stub
    
    			Text line = new Text();
    			List<String> fields = new ArrayList<String>();
    			int size = values.size();
    			for(int i = 0; i < size; i++){
    				BytesRefWritable value = values.get(i);
    				line.set(value.getData(), value.getStart(), value.getLength());
    				fields.add(line.toString());
    			}
    			
    			String rowKey = fields.get(0);
    			String columnFamily = "cf";
    			int length = fieldName.length;
    			ImmutableBytesWritable hKey = new ImmutableBytesWritable();
    			hKey.set(rowKey.getBytes());
    			KeyValue kv = null;
    			for(int i = 1; i < length; i++){
                    kv = new KeyValue(hKey.get(), columnFamily.getBytes(), fieldName[i].getBytes(), fields.get(i).getBytes());
                    context.write(hKey, kv);
    			}
    			
    		}
    	}
    
    }

**注意：**

1.  **生成HFile时，连接的HBase表必须存在。**
2.  **连接的HBase表，可以与最后Bulkload的目标HBase表可以不同名，但是表的结构必须相同（如，Column Family的个数）。**  
    
3.  **版本之间的匹配。**

完整的代码托管在GitHub中： [https://github.com/GatsbyNewton/hive-bulkload-hbase][https_github.com_GatsbyNewton_hive-bulkload-hbase]

[HBaseIntegration]: http://blog.csdn.net/u010376788/article/details/50905476
[distcp]: http://hadoop.apache.org/docs/current/hadoop-distcp/DistCp.html
[MapReduce_RCFile]: http://blog.csdn.net/u010376788/article/details/51051591
[https_github.com_GatsbyNewton_hive-bulkload-hbase]: https://github.com/GatsbyNewton/hive-bulkload-hbase

发表评论取消回复

表情：

评论列表（有 0 条评论，226人围观）

还没有评论，来说两句吧...

相关阅读

相关修炼离线：（四）创建hbase映射表到hive

create database ods; -- drop table if exists ods.ods_yyy_hb; create externa

男娘i/ 2024年03月02日 03:33/ 0 赞/ 60 阅读

相关使用BulkLoading将数据导入HBase

[https://www.jianshu.com/p/7e05a223c18c][https_www.jianshu.com_p_7e05a223c18c] [https:/

约定不等于承诺〃/ 2023年06月13日 04:46/ 0 赞/ 29 阅读

相关 hive数据导入到hbase

方式一： hbase中建表，然后hive中建外部表，这样当hive中写入数据后，hbase中的表也会同时更新创建hbase表 create 'clas

本是古典何须时尚/ 2022年12月10日 15:57/ 0 赞/ 223 阅读

相关 MapReduce读取hive数据，然后用hbase的bulkload方法导入hbase中

废话不多说，直接上代码————————————————————————————————————— 因为默认是一个分区，就只有一个reduce过程，写入比较慢。所以先进

约定不等于承诺〃/ 2022年12月03日 12:55/ 0 赞/ 242 阅读

相关 Bulkload Hive表到HBase

1.描述 HBase可以随机读写海量的数据，但是如果把这海量数据导入到HBase却是一个挑战。如，将Hive表尽可能快的导入到HBase中。这里有以下三种解决方案： 1

水深无声/ 2022年07月26日 00:06/ 0 赞/ 227 阅读

相关 HBase的bulkload使用方法

1.简介　　将数据插入HBase表中的方法很多，我们可以通过TableOutputFormat以Mapreduce on HBase的方式将数据插入，也可以单纯的使用客户端

川长思鸟来/ 2022年06月08日 08:25/ 0 赞/ 523 阅读

相关 HBase数据快速导入之ImportTsv&Bulkload

[HBase数据快速导入之ImportTsv&Bulkload][HBase_ImportTsv_Bulkload] 导入数据最快的方式，可以略过WAL直接生产底层HFi

Myth丶恋晨/ 2022年06月07日 13:53/ 0 赞/ 337 阅读

相关 hbase表创建hive关联表

hbase如果需要到hive里面查询的话，需要建立关联关系表才能在hive里面使用hql进行查询。 hbase表创建hive关联表 CREATE EXTER

朴灿烈づ我的快乐病毒、/ 2022年03月02日 04:22/ 0 赞/ 396 阅读

相关使用Spark通过Bulkload的方式导数据到Hbase

文章目录 [Spark通过Bulk Load 写入Hbase][Spark_Bulk Load _Hbase] [背景][Link 1]

Bertha 。/ 2021年11月02日 07:36/ 0 赞/ 496 阅读

相关 hive与hbase关联表

hive创建关联hbase表有2种形式：一、建立hive内表，指向hbase：（数据是存在hbase中的）通过下面的方式创建hive的内表，这种情况是hbase本来没

小咪咪/ 2021年06月24日 14:37/ 0 赞/ 680 阅读