1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.hadoop.hbase.spark.example.hbasecontext;
18
19 import java.util.ArrayList;
20 import java.util.List;
21
22 import org.apache.hadoop.conf.Configuration;
23 import org.apache.hadoop.hbase.HBaseConfiguration;
24 import org.apache.hadoop.hbase.TableName;
25 import org.apache.hadoop.hbase.client.Put;
26 import org.apache.hadoop.hbase.spark.JavaHBaseContext;
27 import org.apache.hadoop.hbase.util.Bytes;
28 import org.apache.spark.SparkConf;
29 import org.apache.spark.api.java.JavaRDD;
30 import org.apache.spark.api.java.JavaSparkContext;
31 import org.apache.spark.api.java.function.Function;
32
33
34
35
36
37 final public class JavaHBaseBulkPutExample {
38
39 private JavaHBaseBulkPutExample() {}
40
41 public static void main(String[] args) {
42 if (args.length < 2) {
43 System.out.println("JavaHBaseBulkPutExample " +
44 "{tableName} {columnFamily}");
45 return;
46 }
47
48 String tableName = args[0];
49 String columnFamily = args[1];
50
51 SparkConf sparkConf = new SparkConf().setAppName("JavaHBaseBulkPutExample " + tableName);
52 JavaSparkContext jsc = new JavaSparkContext(sparkConf);
53
54 try {
55 List<String> list = new ArrayList<>();
56 list.add("1," + columnFamily + ",a,1");
57 list.add("2," + columnFamily + ",a,2");
58 list.add("3," + columnFamily + ",a,3");
59 list.add("4," + columnFamily + ",a,4");
60 list.add("5," + columnFamily + ",a,5");
61
62 JavaRDD<String> rdd = jsc.parallelize(list);
63
64 Configuration conf = HBaseConfiguration.create();
65
66 JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
67
68 hbaseContext.bulkPut(rdd,
69 TableName.valueOf(tableName),
70 new PutFunction());
71 } finally {
72 jsc.stop();
73 }
74 }
75
76 public static class PutFunction implements Function<String, Put> {
77
78 private static final long serialVersionUID = 1L;
79
80 public Put call(String v) throws Exception {
81 String[] cells = v.split(",");
82 Put put = new Put(Bytes.toBytes(cells[0]));
83
84 put.addColumn(Bytes.toBytes(cells[1]), Bytes.toBytes(cells[2]),
85 Bytes.toBytes(cells[3]));
86 return put;
87 }
88
89 }
90 }