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

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

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

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

HBase数据写路径

（图来自Cloudera）

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

原理

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

其流程如下图：

（图来自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")));}
}


package hbase_mr;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.junit.Test;import java.io.IOException;public class HbaseBulkLoad {static class MyMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put>{private  static final String  COLUMNNAME1="name";private  static final String  COLUMNNAME2="age";private  static final String  FAMILYNAME="f1";@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String[] split = value.toString().split(",");String rowKey=split[0];String name_value = split[1].split(":")[1];String age_value=split[2].split(":")[1];Put p=new Put(Bytes.toBytes(rowKey));p.addColumn(FAMILYNAME.getBytes(),COLUMNNAME1.getBytes(),name_value.getBytes());p.addColumn(FAMILYNAME.getBytes(),COLUMNNAME2.getBytes(),age_value.getBytes());context.write(new ImmutableBytesWritable(rowKey.getBytes()),p);}}public static void main(String[] args) throws  Exception{Configuration conf=new Configuration();Connection conn= ConnectionFactory.createConnection(conf);Table table = conn.getTable(TableName.valueOf("people"));Admin admin = conn.getAdmin();String input = args[0];String output = args[1];Path inPath = new Path(input);Path outPath = new Path(output);Job job=Job.getInstance(conf,"Bulkload");job.setJarByClass(HbaseBulkLoad.class);job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(ImmutableBytesWritable.class);job.setMapOutputValueClass(Put.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(HFileOutputFormat2.class);FileSystem fs=FileSystem.get(conf);if(fs.exists(outPath)){fs.delete(outPath);}FileInputFormat.setInputPaths(job,inPath);FileOutputFormat.setOutputPath(job,outPath);HFileOutputFormat2.configureIncrementalLoad(job,table,conn.getRegionLocator(TableName.valueOf("people")));boolean b = job.waitForCompletion(true);if(b){LoadIncrementalHFiles loadIncrementalHFiles = new LoadIncrementalHFiles(conf);loadIncrementalHFiles.doBulkLoad(outPath,admin,table,conn.getRegionLocator(TableName.valueOf("people")));}System.exit(b?0:1);}}

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

优点：

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

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

3.节省内存

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

php将excel数据导入mysql表中_利用php将xls表格数据以原生方式快速导入mysql
这里以联通的市区编码表为例,如图: 第一步,我们把xls表格中的首行作为mysql表中的字段,为了方便查看这里转换成了小写,表名随意,主键id可加可不加,看自己需求第二步,将xls表格中除了首行以外 ...
mysql快速导入5000万条数据过程记录(LOAD DATA INFILE方式)
mysql快速导入5000万条数据过程记录(LOAD DATA INFILE方式) 首先将要导入的数据文件top5000W.txt放入到数据库数据目录/var/local/mysql/data/${d ...
java导入包大全_eclipse快速导入jar包的相关操作步骤
eclipse怎样快速导入jar包呢?熟悉这款软件是非常简单的,今天小编就分享了关于eclipse快速导入jar包,有需要的朋友一起来看看吧! eclipse快速导入jar包的相关操作步骤方法1·最 ...
关系型数据库到HBase的数据储存方式变迁
如今Bigtable型(列族)数据库应用越来越广,功能也很强大.但是很多人还是把它当做关系型数据库在使用,用原来关系型数据库的思维建表.存储.查询.本文以hbase举例讲述数据模式的变化. 传统关系型 ...
关于Maven中的常见命令，通过命令的方式快速创建一个空的maven工程，将jar包打到maven仓库中
打jar包的方式 mvn package -Dmaven.test.skip=true mvn install -Dmaven.test.skip=true mvn compile -Dmave ...
android studio 自动生成对象,在Android Studio中快速导入可绘制对象的方法？
在Android Studio中快速导入可绘制对象的方法? 在Android项目中的某个时候,您将需要在res/drawable目录中导入一些可绘制对象-工具栏图标,图像,UI功能. 大多数情况下,您 ...
mysql 快速导出_mysql 快速导入导出
随着数据库的数据越来越大,采用mysqldump 越来越慢,测试环境的机器配置不高,2G左右的数据导入进入像蜗牛一般,非常影响效率,这里采用一些改进的方法来比以前导入的速度提高好几倍,但日常配备应有更 ...
HBase的BulkLoad详解
HBase的BulkLoad主要分为HFile生成和HFile导入两个阶段. HFile生成阶段 HFile导入阶段
问卷星问卷数据怎么快速导入SPSSAU？
最近收到小伙伴询问问卷星导入的问卷数据怎么编码? 现在的问卷调查,很多都是通过网络问卷的方式进行,问卷星是一个专业的在线问卷调查.测评投票平台,如果你的问卷正好是在问卷星网站发放,填答,回收数据,那太 ...

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

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

最新文章

热门文章