大数据实战（上）-HBase Bulkload实例

大纲

简介bulkload

演示编写bulkload实例

Bulkload

- 适用于大量数据，一次性导入hbase

- 演示编写bulkload实例

处理文本

a a
b b
c c
d d
e e
f f

实例分析

代码编写：

建立java项目，导入hbase相关jar包，建立BulkloadMR.java类

package com.song.bulkload;
public class BulkloadMR {
public static class BulkM extends Mapper<Object, Text, ImmutableBytesWritable, Put> {
public final static String SP = "\t";

protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
// value切分
String[] values = value.toString().split(SP);
if (values.length == 2) {
byte[] rowkey = Bytes.toBytes(values[0]);
byte[] c_v = Bytes.toBytes(values[1]);
byte[] family = Bytes.toBytes("d");
byte[] cloumn = Bytes.toBytes("c");
ImmutableBytesWritable rowkeyWriable = new ImmutableBytesWritable(rowkey);
Put put = new Put(rowkey);
put.add(family, cloumn, c_v);
Context.write(rowkeyWriable, put);
}
}
}

public static void main(String[] args) throws IOException {
// args0 dst
// args1 out
// args2 split MB
// args3 hbase table name
if (args.length != 4) {
System.exit(0);
}
String dst = args[0];
String out = args[1];
int SplitMB = Integer.valueOf(args[2]);
String table_name = args[3];
// 设置属性对应参数
Configuration conf = new Configuration();
conf.set("mapreduce.input.fileinputformat.split.maxsize", String.valueOf(SplitMB * 1024 * 1024));
conf.set("mapred.min.split.size", String.valueOf(SplitMB * 1024 * 1024));
conf.set("mapreduce.input.fileinputformat.split.minsize.per.node", String.valueOf(SplitMB * 1024 * 1024));
conf.set("mapreduce.input.fileinputformat.split.minsize.per.rack", String.valueOf(SplitMB * 1024 * 1024));
Job job = new Job(conf, "BulkLoad");
job.setJarByClass(BulkloadMR.class);
job.setMapperClass(BulkM.class);
job.setReducerClass(PutSortReducer.class);
job.setOutputFormatClass(HFileOutputFormat.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);
FileInputFormat.addInputPath(job, new Path(dst));
FileOutPutFormat.setOutputPath(job, new Path(out));
// MapReudce 初始化一张表
Configuration hbaseConf = HBaseConfiguration.create();
HTable table = new HTable(hbaseConf, table_name);
HFileOutputFormat.configurementailLoad(job, table);
//执行job任务
job.waitForCompletion(true);
//将HFile文件导入HBase
LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
loader.doBulkLoad(new Path(out), table);
}
}

上面代码首先将HDFS中的数据文件通过MapReduce任务生成HFile文件，然后将HFile文件导入HBase数据表（该数据表已存在）。

设置列族属性，才能正确导入数据，那如何动态设置列组，或者每个列族对应不同文件来划分。