Hbase调用JavaAPI实现批量导入操作

将手机上网日志文件批量导入到Hbase中。操作步骤：

1、将日志文件(请下载附件)上传到HDFS中，利用hadoop的操作命令上传：hadoop  fs -put input  /

2、创建Hbase表，通过Java操作

Java代码  

1. package com.jiewen.hbase;

2. import java.io.IOException;

3. import org.apache.hadoop.conf.Configuration;

4. import org.apache.hadoop.hbase.HBaseConfiguration;

5. import org.apache.hadoop.hbase.HColumnDescriptor;

6. import org.apache.hadoop.hbase.HTableDescriptor;

7. import org.apache.hadoop.hbase.client.Get;

8. import org.apache.hadoop.hbase.client.HBaseAdmin;

9. import org.apache.hadoop.hbase.client.HTable;

10. import org.apache.hadoop.hbase.client.Put;

11. import org.apache.hadoop.hbase.client.Result;

12. import org.apache.hadoop.hbase.client.ResultScanner;

13. import org.apache.hadoop.hbase.client.Scan;

14. import org.apache.hadoop.hbase.util.Bytes;

15. public class HbaseDemo {

16. public static void main(String[] args) throws IOException {

17. String tableName = "wlan_log";

18. String columnFamily = "cf";

19. HbaseDemo.create(tableName, columnFamily);

20. // HbaseDemo.put(tableName, "row1", columnFamily, "cl1", "data");

21. // HbaseDemo.get(tableName, "row1");

22. // HbaseDemo.scan(tableName);

23. // HbaseDemo.delete(tableName);

24. }

25. // hbase操作必备

26. private static Configuration getConfiguration() {

27. Configuration conf = HBaseConfiguration.create();

28. conf.set("hbase.rootdir", "hdfs://hadoop1:9000/hbase");

29. // 使用eclipse时必须加入这个，否则无法定位

30. conf.set("hbase.zookeeper.quorum", "hadoop1");

31. return conf;

32. }

33. // 创建一张表

34. public static void create(String tableName, String columnFamily)

35. throws IOException {

36. HBaseAdmin admin = new HBaseAdmin(getConfiguration());

37. if (admin.tableExists(tableName)) {

38. System.out.println("table exists!");

39. } else {

40. HTableDescriptor tableDesc = new HTableDescriptor(tableName);

41. tableDesc.addFamily(new HColumnDescriptor(columnFamily));

42. admin.createTable(tableDesc);

43. System.out.println("create table success!");

44. }

45. }

46. // 加入一条记录

47. public static void put(String tableName, String row, String columnFamily,

48. String column, String data) throws IOException {

49. HTable table = new HTable(getConfiguration(), tableName);

50. Put p1 = new Put(Bytes.toBytes(row));

51. p1.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes

52. .toBytes(data));

53. table.put(p1);

54. System.out.println("put'" + row + "'," + columnFamily + ":" + column

55. + "','" + data + "'");

56. }

57. // 读取一条记录

58. public static void get(String tableName, String row) throws IOException {

59. HTable table = new HTable(getConfiguration(), tableName);

60. Get get = new Get(Bytes.toBytes(row));

61. Result result = table.get(get);

62. System.out.println("Get: " + result);

63. }

64. // 显示全部数据

65. public static void scan(String tableName) throws IOException {

66. HTable table = new HTable(getConfiguration(), tableName);

67. Scan scan = new Scan();

68. ResultScanner scanner = table.getScanner(scan);

69. for (Result result : scanner) {

70. System.out.println("Scan: " + result);

71. }

72. }

73. // 删除表

74. public static void delete(String tableName) throws IOException {

75. HBaseAdmin admin = new HBaseAdmin(getConfiguration());

76. if (admin.tableExists(tableName)) {

77. try {

78. admin.disableTable(tableName);

79. admin.deleteTable(tableName);

80. } catch (IOException e) {

81. e.printStackTrace();

82. System.out.println("Delete " + tableName + " 失败");

83. }

84. }

85. System.out.println("Delete " + tableName + " 成功");

86. }

87. }

3、将日志文件导入Hbase表wlan_log中：

Java代码  

1. import java.text.SimpleDateFormat;

2. import java.util.Date;

3. import org.apache.hadoop.conf.Configuration;

4. import org.apache.hadoop.hbase.client.Put;

5. import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;

6. import org.apache.hadoop.hbase.mapreduce.TableReducer;

7. import org.apache.hadoop.hbase.util.Bytes;

8. import org.apache.hadoop.io.LongWritable;

9. import org.apache.hadoop.io.NullWritable;

10. import org.apache.hadoop.io.Text;

11. import org.apache.hadoop.mapreduce.Counter;

12. import org.apache.hadoop.mapreduce.Job;

13. import org.apache.hadoop.mapreduce.Mapper;

14. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

15. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

16. public class HbaseBatchImport {

17. public static void main(String[] args) throws Exception {

18. final Configuration configuration = new Configuration();

19. // 设置zookeeper

20. configuration.set("hbase.zookeeper.quorum", "hadoop1");

21. // 设置hbase表名称

22. configuration.set(TableOutputFormat.OUTPUT_TABLE, "wlan_log");

23. // 将该值改大，防止hbase超时退出

24. configuration.set("dfs.socket.timeout", "180000");

25. final Job job = new Job(configuration, "HBaseBatchImport");

26. job.setMapperClass(BatchImportMapper.class);

27. job.setReducerClass(BatchImportReducer.class);

28. // 设置map的输出，不设置reduce的输出类型

29. job.setMapOutputKeyClass(LongWritable.class);

30. job.setMapOutputValueClass(Text.class);

31. job.setInputFormatClass(TextInputFormat.class);

32. // 不再设置输出路径。而是设置输出格式类型

33. job.setOutputFormatClass(TableOutputFormat.class);

34. FileInputFormat.setInputPaths(job, "hdfs://hadoop1:9000/input");

35. job.waitForCompletion(true);

36. }

37. static class BatchImportMapper extends

38. Mapper<LongWritable, Text, LongWritable, Text> {

39. SimpleDateFormat dateformat1 = new SimpleDateFormat("yyyyMMddHHmmss");

40. Text v2 = new Text();

41. protected void map(LongWritable key, Text value, Context context)

42. throws java.io.IOException, InterruptedException {

43. final String[] splited = value.toString().split("\t");

44. try {

45. final Date date = new Date(Long.parseLong(splited[0].trim()));

46. final String dateFormat = dateformat1.format(date);

47. String rowKey = splited[1] + ":" + dateFormat;

48. v2.set(rowKey + "\t" + value.toString());

49. context.write(key, v2);

50. } catch (NumberFormatException e) {

51. final Counter counter = context.getCounter("BatchImport",

52. "ErrorFormat");

53. counter.increment(1L);

54. System.out.println("出错了" + splited[0] + " " + e.getMessage());

55. }

56. };

57. }

58. static class BatchImportReducer extends

59. TableReducer<LongWritable, Text, NullWritable> {

60. protected void reduce(LongWritable key,

61. java.lang.Iterable<Text> values, Context context)

62. throws java.io.IOException, InterruptedException {

63. for (Text text : values) {

64. final String[] splited = text.toString().split("\t");

65. final Put put = new Put(Bytes.toBytes(splited[0]));

66. put.add(Bytes.toBytes("cf"), Bytes.toBytes("date"), Bytes

67. .toBytes(splited[1]));

68. // 省略其它字段，调用put.add(....)就可以

69. context.write(NullWritable.get(), put);

70. }

71. };

72. }

73. }

4、查看导入结果：