通过BulkLoad的方式快速导入海量数据

系统管理员 2022-01-13 05:53 241阅读 0赞

### **摘要** ###

加载数据到HBase的方式有多种，通过HBase API导入或命令行导入或使用第三方(如sqoop)来导入或使用MR来批量导入（耗费磁盘I/O，容易在导入的过程使节点宕机），但是这些方式不是慢就是在导入的过程的占用Region资料导致效率低下，今天要讲的就是利用HBase在HDFS存储原理及MapReduce的特性来快速导入海量的数据

### **HBase数据在HDFS下是如何存储的？** ###

HBase中每张Table在根目录（/HBase）下用一个文件夹存储，Table名为文件夹名，在Table文件夹下每个Region同样用一个文件夹存储，每个Region文件夹下的每个列族也用文件夹存储，而每个列族下存储的就是一些HFile文件，HFile就是HBase数据在HFDS下存储格式，其整体目录结构如下：

**/hbase/<tablename>/<encoded-regionname>/<column-family>/<filename>**

### **HBase数据写路径** ###

![776259-20160604212844274-56299948.png][]

（图来自Cloudera）

在put数据时会先将数据的更新操作信息和数据信息写入WAL，在写入到WAL后，数据就会被放到MemStore中，当MemStore满后数据就会被flush到磁盘(即形成HFile文件),在这过程涉及到的flush,split,compaction等操作都容易造成节点不稳定，数据导入慢，耗费资源等问题，在海量数据的导入过程极大的消耗了系统性能，避免这些问题最好的方法就是使用BlukLoad的方式来加载数据到HBase中。

### **原理** ###

**利用HBase数据按照HFile格式存储在HDFS的原理，使用Mapreduce直接生成HFile格式文件后，RegionServers再将HFile文件移动到相应的Region目录下**

其流程如下图：

![776259-20160604212935164-1040085823.png][]

（图来自Cloudera）

### **导入过程** ###

**1.使用MapReduce生成HFile文件**

**GenerateHFile类**

public class GenerateHFile extends Mapper<LongWritable,
            Text, ImmutableBytesWritable, Put>{
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
                String line = value.toString();
                String[] items = line.split("\t");
     
                String ROWKEY = items[1] + items[2] + items[3];
                ImmutableBytesWritable rowkey = new ImmutableBytesWritable(ROWKEY.getBytes());
                Put put = new Put(ROWKEY.getBytes());   //ROWKEY
                put.addColumn("INFO".getBytes(), "URL".getBytes(), items[0].getBytes());
                put.addColumn("INFO".getBytes(), "SP".getBytes(), items[1].getBytes());  //出发点
                put.addColumn("INFO".getBytes(), "EP".getBytes(), items[2].getBytes());  //目的地
                put.addColumn("INFO".getBytes(), "ST".getBytes(), items[3].getBytes());   //出发时间
                put.addColumn("INFO".getBytes(), "PRICE".getBytes(), Bytes.toBytes(Integer.valueOf(items[4])));  //价格
                put.addColumn("INFO".getBytes(), "TRAFFIC".getBytes(), items[5].getBytes());//交通方式
                put.addColumn("INFO".getBytes(), "HOTEL".getBytes(), items[6].getBytes()); //酒店
               
                context.write(rowkey, put);
            }
    }

**GenerateHFileMain类**

public class GenerateHFileMain {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            final String INPUT_PATH= "hdfs://master:9000/INFO/Input";
            final String OUTPUT_PATH= "hdfs://master:9000/HFILE/Output";
            Configuration conf = HBaseConfiguration.create();
            Connection connection = ConnectionFactory.createConnection(conf);
            Table table = connection.getTable(TableName.valueOf("TRAVEL"));
            Job job=Job.getInstance(conf);
            job.getConfiguration().set("mapred.jar", "/home/hadoop/TravelProject/out/artifacts/Travel/Travel.jar");  //预先将程序打包再将jar分发到集群上
            job.setJarByClass(GenerateHFileMain.class);
            job.setMapperClass(GenerateHFile.class);
            job.setMapOutputKeyClass(ImmutableBytesWritable.class);
            job.setMapOutputValueClass(Put.class);
            job.setOutputFormatClass(HFileOutputFormat2.class);
            HFileOutputFormat2.configureIncrementalLoad(job,table,connection.getRegionLocator(TableName.valueOf("TRAVEL")))
            FileInputFormat.addInputPath(job,new Path(INPUT_PATH));
            FileOutputFormat.setOutputPath(job,new Path(OUTPUT_PATH));
            System.exit(job.waitForCompletion(true)?0:1);
        }

**注意**

1.Mapper的输出Key类型必须是包含Rowkey的ImmutableBytesWritable格式，Value类型必须为KeyValue或Put类型，当导入的数据有多列时使用Put，只有一个列时使用KeyValue

2.job.setMapOutPutValueClass的值决定了job.setReduceClass的值，这里Reduce主要起到了对数据进行排序的作用，当job.setMapOutPutValueClass的值Put.class和KeyValue.class分别对应job.setReduceClass的PutSortReducer和KeyValueSortReducer

3.在创建表时对表进行预分区再结合MapReduce的并行计算机制能加快HFile文件的生成，如果对表进行了预分区(Region)就设置Reduce数等于分区数（Region）

4.在多列族的情况下需要进行多次的context.write

**2.通过BlukLoad方式加载HFile文件**

public class LoadIncrementalHFileToHBase {
        public static void main(String[] args) throws Exception {
            Connection connection = ConnectionFactory.createConnection(conf);
            Admin admin = connection.getAdmin();
            Table table = connection.getTable(TableName.valueOf("TRAVEL"));
            LoadIncrementalHFiles load = new LoadIncrementalHFiles(conf);
            load.doBulkLoad(new Path("hdfs://master:9000/HFILE/OutPut"), admin,table,connection.getRegionLocator(TableName.valueOf("TRAVEL")));
        }
    }

由于BulkLoad是绕过了Write to WAL，Write to MemStore及Flush to disk的过程，所以并不能通过WAL来进行一些复制数据的操作

**优点：**

1.导入过程不占用Region资源

2.能快速导入海量的数据

3.节省内存

[776259-20160604212844274-56299948.png]: /images/20220113/ce5dc661149648eeb541c71e08bcda02.png
[776259-20160604212935164-1040085823.png]: /images/20220113/d28c13c257c840ccb9a6388dc9bb43a8.png

发表评论取消回复

表情：

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

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

相关阅读

相关 mysql快速导入数据方式

mysql快速导入数据方式 1、描述：导入大量数据时，如何快速导入数据，可通过修改下列参数的方式成倍提高速度 1.1 innodb\_flush\_lo

左手的ㄟ右手/ 2022年12月22日 09:59/ 0 赞/ 172 阅读

相关 HBase ：利用SQL BulkLoad快速导入海量数据

HBase BulkLoad概述直接用put命令将海量数据导入至HBase会耗费大量时间（HFile合并、Region分割等），而且用原生的put命令导入海量数据，会占

﹏ヽ暗。殇╰゛Y/ 2022年10月02日 10:44/ 0 赞/ 211 阅读

相关通过BlukLoad的方式快速导入海量数据

[通过BlukLoad的方式快速导入海量数据][BlukLoad] http://www.cnblogs.com/MOBIN/p/5559575.html 摘要加载

女爷i/ 2022年09月23日 14:58/ 0 赞/ 137 阅读

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

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

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

相关 HBase系列(三)：利用SQL BulkLoad快速导入海量数据

HBase BulkLoad概述直接用put命令将海量数据导入至HBase会耗费大量时间（HFile合并、Region分割等），而且用原生的put命令导入海量数据，会占

柔光的暖阳◎/ 2022年05月19日 06:06/ 0 赞/ 228 阅读

相关 Sphinx 实现海量数据的快速查询

来源：[https://blog.csdn.net/qq\_28018283/article/details/72622003][https_blog.csdn.net_qq_

客官°小女子只卖身不卖艺/ 2022年05月19日 00:37/ 0 赞/ 195 阅读

相关 HBase系列：利用BulkLoad快速导入海量数据及集群间数据同步

Region的大小单个region最大大小官方推荐5~10GB，这是三备份前的数据大小，通过hbase.hregion.max.filesize配置，当超过这个值后regi

红太狼/ 2022年05月17日 09:24/ 0 赞/ 998 阅读

相关 Hbase通过BulkLoad的方式快速导入海量数据

HBase数据在HDFS下是如何存储的？ HBase中每张Table在根目录（/HBase）下用一个文件夹存储，Table名为文件夹名，在Table文件夹下每个Region同

Dear 丶/ 2022年05月17日 09:24/ 0 赞/ 197 阅读

相关通过BulkLoad的方式快速导入海量数据

摘要加载数据到HBase的方式有多种，通过HBase API导入或命令行导入或使用第三方(如sqoop)来导入或使用MR来批量导入（耗费磁盘I/O，容易在导入的过程使节

系统管理员/ 2022年01月13日 05:53/ 0 赞/ 242 阅读

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

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

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