HBase导入大数据三大方式之（三）——mapreduce+completebulkload 方式


做大数据时，经常需要用到将大量格式化的文本数据导入到hbase中。此处就用到的三种方式：hive类SQL语句方式、importtsv +completebulkload 方式、mapreduce+completebulkload 方式，做下简单示例。hive类SQL语句方式和importtsv +completebulkload 方式都已经介绍过了，现在介绍mapreduce+completebulkload 方式：

实例中，我以虚拟话单作为需要导入的数据，格式上上篇文章中有介绍。

一、设计HBASE的表名、列名和字段名：

1、  创建支持hive查询的hbase表：

CREATE TABLE bill(rowkey string, 00_selfnumber string, 01_day string, 02_hour string, 03_duration string, 04_calltype string, 05_targetnumber string, 06_address string, 07_longdtype string, 08_basecost float, 09_longdcostfloat, 10_infocostfloat,11_privilege string) STORED BY'org.apache.hadoop.hive.hbase.HBaseStorageHandler' WITH SERDEPROPERTIES("hbase.columns.mapping"="info:00_selfnumber,info:01_day,info:02_hour,info:03_duration,info:04_calltype,info:05_targetnumber,info:06_address,info:07_longdtype,info:08_basecost,info:09_longdcost,info:10_infocost,info:11_privilege") TBLPROPERTIES ("hbase.table.name"="bill");

说明：

1、确定了表名：bill；列名：info；字段名：01_day~11_privilege。

2、第一排的”bill”是在hive中的表名，最后一排的”bill”是对应hbase中的表名。两个表名可以不同，字段名需要相同。

注意：目前导入方法只支持一个列族，这里是“info”。Hbase中的字段名必须要按照alphabet人为排好序。

二、上传文件到HDFS：

1、  方法一：使用eclipse插件：



2、  方法二：put命令：

hadoop@master:/usr/hadoop/hadoop-2.0.0-cdh4.2.2$ bin/hdfs dfs-put bill.txt/user/hadoop/loadDataToHBase/input


1、  其它，如hadoopAPI，调用OTL可执行程序。

略。

说明：这里确定上传到HDFS的input文件的路径，如：

/user/hadoop/loadDataToHBase/input

三、写mapreduce代码将需要导入的数据生成hfile的格式：

LoadDataToHBase.java

package com.melina;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Iterator;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
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.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
import org.apache.hadoop.hbase.mapreduce.KeyValueSortReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class LoadDataToHBase {
	
	public static class LoadDataToHBaseMapper extends
			Mapper<LongWritable, Text, ImmutableBytesWritable, Text> {	
		
		public static int y,m,d;   
		Calendar cal=Calendar.getInstance();    
		//map的key用一个immutableBytesWritable类型的无意义的key，map的value是直接将原来的一行记录输出，
		//map完成后会shuffle和sort，将key-value按照key排序，否则写不进hfile，hfile要求后写的key不能小于先写的key
		private ImmutableBytesWritable immutableBytesWritable = new ImmutableBytesWritable();
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			immutableBytesWritable.set(Bytes.toBytes(key.get())); 
	        context.write(immutableBytesWritable, value);
		}
	}
	
	//reducer每次得到map输出的value就是输入文件的中的一行
	//reducer的key也无意义，每一个value值就是一个hfile格式的输出，包括rowkey、family、qualifier、timestamp、value
	//其中rowkey是由“年月日+该行的行号”组成，每行的行号由预处理的时候加在每一行的开头
	public static class LoadDataToHBaseReducer extends
			Reducer<ImmutableBytesWritable, Text, ImmutableBytesWritable, KeyValue> {

		public static int y, m, d;
		Calendar cal = Calendar.getInstance();

		protected void reduce(ImmutableBytesWritable key, Iterable<Text> values, 
	            Context context)
				throws IOException, InterruptedException {
			String value="";       
	        while(values.iterator().hasNext()) 
	        { 
	            value = values.iterator().next().toString(); 
	            if(value != null && !"".equals(value)) 
	            { 
	            	List<KeyValue> list = new ArrayList<KeyValue>();
	            	list = createKeyValue(value.toString());
	            	Iterator<KeyValue> it = list.iterator();
	        		while (it.hasNext()) {
	        			KeyValue kv = new KeyValue();
	        			kv = it.next();
	        			if(kv!=null) {
	        				context.write(key, kv);
	        			}                       
	        		}
	            } 
	        }
	    } 

		private List<KeyValue> createKeyValue(String str) {
			List<KeyValue> list = new ArrayList<KeyValue>();
				String[] values = str.toString().split(",");
				String[] qualifiersName = CONSTANT.qualifiersName;
				//long timeStamp = CONSTANT.timeStamp;
				for (int i = 1; i < qualifiersName.length; i++) {
					String rownum = values[0];
					String family = CONSTANT.familyName;
					String qualifier = qualifiersName[i];
					String value_str = values[i];
					y=cal.get(Calendar.YEAR);    
					m=cal.get(Calendar.MONTH);    
					d=cal.get(Calendar.DATE); 
					String rowkey_str = Integer.toString(y)+Integer.toString(m)+Integer.toString(d)+rownum;
					KeyValue kv = new KeyValue(Bytes.toBytes(rowkey_str),
							Bytes.toBytes(family), Bytes.toBytes(qualifier),
							System.currentTimeMillis(), Bytes.toBytes(value_str));
				list.add(kv);
			}
			return list;
		}

		
}

	public static void main(String[] args) throws IOException,
			InterruptedException, ClassNotFoundException {
		Configuration conf = HBaseConfiguration.create();
		Job job = new Job(conf, CONSTANT.jobName);
		job.setJarByClass(LoadDataToHBase.class);

		job.setOutputKeyClass(ImmutableBytesWritable.class);
		//注意此处的Text.class要与map函数的输出key-value的value类型相对应
		job.setOutputValueClass(Text.class);

		job.setMapperClass(LoadDataToHBaseMapper.class);
		job.setReducerClass(LoadDataToHBaseReducer.class);
		// job.setOutputFormatClass(org.apache.hadoop.hbase.mapreduce.HFileOutputFormat.class);
		job.setOutputFormatClass(HFileOutputFormat.class);
		// job.setNumReduceTasks(4);
		// job.setPartitionerClass(org.apache.hadoop.hbase.mapreduce.SimpleTotalOrderPartitioner.class);

		Configuration fs_conf = new Configuration();
		FileSystem fs = FileSystem.get(fs_conf);
		String str_inPath = CONSTANT.str_inPath;
		String str_outPath = CONSTANT.str_outPath;
		//如果输出路径存在就先删掉，因为不允许输出路径事先存在
		Path outPath = new Path(str_outPath);
		if (fs.exists(outPath))
			fs.delete(outPath, true);

		FileInputFormat.addInputPath(job, new Path(str_inPath));
		FileOutputFormat.setOutputPath(job, new Path(str_outPath));

		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}

我把需要用到的一些路径参数之类的写到了CONSTANT.java 

package com.melina;

public class CONSTANT {
	public static final String jobName = "LoadDataToHBase";
	public static final String[] qualifiersName = { "","00_selfnumber", "01_day", "02_hour", "03_duration",
			"04_calltype", "05_targetnumber", "06_address", "07_longdtype",
			"08_basecost", "09_longdcost", "10_infocost", "11_privilege" };
	public static final String familyName = "info";
	public static final String tableName = "bill";
	public static final String str_inPath = "/user/hadoop/loadDataToHBase/input";
	public static final String str_outPath = "/user/hadoop/loadDataToHBase/output";
	public static final long timeStamp = System.currentTimeMillis();

}

需要相应地修改CONSTANT：

1、  把上面涉及到的表名、列名、字段名和输入文件的目录写进去。

2、  随便指定一个输出目录。如：

/user/hadoop/loadDataToHBase/output


说明：mapreduce的输出路径不允许实现存在，因为按理说这个输出目录应该为notexist的。不过程序里面会先判断这个路径如果事先存在则删除。所以可以填写一个已经存在的路径并且该路径下的文件可以被删除。

四、编译打包：

1、  java代码打包成jar：


五、修改运行脚本文件：

1、  把上一步打好的jar包放在集群中任意主机任意路径下，如：

/home/hadoop

2、  创建或者修改脚本文件LoadDataToHBase.sh，该脚本文件的作用是执行上面的jar包生成hfile，之后再将hfile导入到第一步建的hbase表中。

#!/bin/bash

/usr/hadoop/hadoop-2.0.0-mr1-cdh4.2.2/bin/hadoop jar /home/hadoop/LoadDataToHBase.jar com.melina.LoadDataToHBase&&

/usr/hadoop/hadoop-2.0.0-mr1-cdh4.2.2/bin/hadoop jar /usr/hadoop/hbase-0.94.2-cdh4.2.2/hbase-0.94.2-cdh4.2.2-security.jar completebulkload /user/hadoop/loadDataToHBase/output bill

说明：

1、  该脚本中包含两条命令，中间用”&&”分开，表示需要等待上一条命令执行完成之后才能执行第二条命令。

2、  第一条命令指明要执行的jar包就是刚刚生成的jar包，以及要执行的main所在的类。

3、  第二条命令要执行的jar包就是hbase的jar包，执行completebulkload后面是hfile在HDFS上的存放路径以及要导的hbase表。

可以看出这个方法与importtsv +completebulkload 的区别在于用mapreduce替代了importtsv将数据生成了hfile格式。


六、运行：

1、  让脚本可执行：

Sudo chmod +x/home/hadoop/LoadDataToHBase.sh

2、  执行脚本：

./LoadDataToHBase.sh

七、查看结果：

1、  通过eclipse可以看到HDFS上面hbase里面的表bill里面已经有hfile了：



2、  通过hive命令查询hbase的表bill的数据：

hive> select *from bill limit 10;


八、附件：

1、  话单预处理之后的数据bill.txt。

2、  Mapreduce程序工程。

3、  Mapreduce程序jar包。

4、  脚本文件LoadDataToHBase.sh。

下载地址：http://download.csdn.net/detail/luckymelina/7142747



OK！GOOD LUCK！小伙伴们加油！

相关内容