Hbase的bulkload代码

 1 BulkLoadDriver.java
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.conf.Configured;
 5 import org.apache.hadoop.fs.FileSystem;
 6 import org.apache.hadoop.fs.Path;
 7 import org.apache.hadoop.hbase.HBaseConfiguration;
 8 import org.apache.hadoop.hbase.TableName;
 9 import org.apache.hadoop.hbase.client.Connection;
10 import org.apache.hadoop.hbase.client.ConnectionFactory;
11 import org.apache.hadoop.hbase.client.Put;
12 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
13 import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
14 import org.apache.hadoop.mapreduce.Job;
15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
16 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
18 import org.apache.hadoop.util.Tool;
19 import org.apache.hadoop.util.ToolRunner;
20 /**
21 * Created by shaobo on 15-6-9.
22 */
23 public class BulkLoadDriver extends Configured implements Tool {
24 private static final String DATA_SEPERATOR = "\\s+";
25 private static final String TABLE_NAME = "temperature";//表名
26 private static final String COLUMN_FAMILY_1="date";//列组1
27 private static final String COLUMN_FAMILY_2="tempPerHour";//列组2
28 public static void main(String[] args) {
29           try {
30                 int response = ToolRunner.run(HBaseConfiguration.create(), new BulkLoadDriver(), args);
31                 if(response == 0) {
32                    System.out.println("Job is successfully completed...");
33                 } else {
34                           System.out.println("Job failed...");
35                          }
36                 } catch(Exception exception) {
37                         exception.printStackTrace();
38                    }
39                 }
40  public int run(String[] args) throws Exception {
41  String outputPath = args[1];
42 /**
43 * 设置作业参数
44 */
45 Configuration configuration = getConf();
46 configuration.set("data.seperator", DATA_SEPERATOR);
47 configuration.set("hbase.table.name", TABLE_NAME);
48 configuration.set("COLUMN_FAMILY_1", COLUMN_FAMILY_1);
49 configuration.set("COLUMN_FAMILY_2", COLUMN_FAMILY_2);
50 Job job = Job.getInstance(configuration, "Bulk Loading HBase Table::" + TABLE_NAME);
51 job.setJarByClass(BulkLoadDriver.class);
52 job.setInputFormatClass(TextInputFormat.class);
53 job.setMapOutputKeyClass(ImmutableBytesWritable.class);//指定输出键类
54 job.setMapOutputValueClass(Put.class);//指定输出值类
55 job.setMapperClass(BulkLoadMapper.class);//指定Map函数
56 FileInputFormat.addInputPaths(job, args[0]);//输入路径
57 FileSystem fs = FileSystem.get(configuration);
58 Path output = new Path(outputPath);
59 if (fs.exists(output)) {
60 fs.delete(output, true);//如果输出路径存在，就将其删除
61 }
62 FileOutputFormat.setOutputPath(job, output);//输出路径
63 Connection connection = ConnectionFactory.createConnection(configuration);
64 TableName tableName = TableName.valueOf(TABLE_NAME);
65 HFileOutputFormat2.configureIncrementalLoad(job, connection.getTable(tableName), connection.getRegionLocator(tableName));
66 job.waitForCompletion(true);
67 if (job.isSuccessful()){
68 HFileLoader.doBulkLoad(outputPath, TABLE_NAME);//导入数据
69 return 0;
70 } else {
71 return 1;
72 }
73 }
74 }

 1 BulkLoadMapper.java
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.hbase.client.Put;
 5 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
 6 import org.apache.hadoop.hbase.util.Bytes;
 7 import org.apache.hadoop.io.LongWritable;
 8 import org.apache.hadoop.io.Text;
 9 import org.apache.hadoop.mapreduce.Mapper;
10 /**
11 * Created by shaobo on 15-6-9.
12 */
13 public class BulkLoadMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
14 private String hbaseTable;
15 private String dataSeperator;
16 private String columnFamily1;
17 private String columnFamily2;
18 public void setup(Context context) {
19 Configuration configuration = context.getConfiguration();//获取作业参数
20 hbaseTable = configuration.get("hbase.table.name");
21 dataSeperator = configuration.get("data.seperator");
22 columnFamily1 = configuration.get("COLUMN_FAMILY_1");
23 columnFamily2 = configuration.get("COLUMN_FAMILY_2");
24 }
25 public void map(LongWritable key, Text value, Context context){
26 try {
27 String[] values = value.toString().split(dataSeperator);
28 ImmutableBytesWritable rowKey = new ImmutableBytesWritable(values[0].getBytes());
29 Put put = new Put(Bytes.toBytes(values[0]));
30 put.addColumn(Bytes.toBytes(columnFamily1), Bytes.toBytes("month"), Bytes.toBytes(values[1]));
31 put.addColumn(Bytes.toBytes(columnFamily1), Bytes.toBytes("day"), Bytes.toBytes(values[2]));
32 for (int i = 3; i < values.length; ++i){
33 put.addColumn(Bytes.toBytes(columnFamily2), Bytes.toBytes("hour : " + i), Bytes.toBytes(values[i]));
34 }
35 context.write(rowKey, put);
36 } catch(Exception exception) {
37 exception.printStackTrace();
38 }
39 }
40 
41 }

 1 HFileLoader.java
 2 
 3 import org.apache.hadoop.conf.Configuration;
 4 import org.apache.hadoop.fs.Path;
 5 import org.apache.hadoop.hbase.HBaseConfiguration;
 6 import org.apache.hadoop.hbase.client.HTable;
 7 import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
 8 /**
 9 * Created by shaobo on 15-6-9.
10 */
11 public class HFileLoader {
12 public static void doBulkLoad(String pathToHFile, String tableName){
13 try {
14 Configuration configuration = new Configuration();
15 HBaseConfiguration.addHbaseResources(configuration);
16 LoadIncrementalHFiles loadFfiles = new LoadIncrementalHFiles(configuration);
17 HTable hTable = new HTable(configuration, tableName);//指定表名
18 loadFfiles.doBulkLoad(new Path(pathToHFile), hTable);//导入数据
19 System.out.println("Bulk Load Completed..");
20 } catch(Exception exception) {
21 exception.printStackTrace();
22 }
23 }
24 
25 }