1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.hadoop.hbase.spark;
18
19 import java.io.File;
20 import java.io.IOException;
21 import java.io.Serializable;
22
23 import java.util.*;
24
25 import org.apache.commons.logging.Log;
26 import org.apache.commons.logging.LogFactory;
27 import org.apache.hadoop.conf.Configuration;
28 import org.apache.hadoop.hbase.Cell;
29 import org.apache.hadoop.hbase.CellUtil;
30 import org.apache.hadoop.hbase.HBaseTestingUtility;
31 import org.apache.hadoop.hbase.TableName;
32 import org.apache.hadoop.hbase.client.Connection;
33 import org.apache.hadoop.hbase.client.ConnectionFactory;
34 import org.apache.hadoop.hbase.client.Delete;
35 import org.apache.hadoop.hbase.client.Get;
36 import org.apache.hadoop.hbase.client.Put;
37 import org.apache.hadoop.hbase.client.Result;
38 import org.apache.hadoop.hbase.client.Scan;
39 import org.apache.hadoop.hbase.client.Table;
40 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
41 import org.apache.hadoop.hbase.spark.example.hbasecontext.JavaHBaseBulkDeleteExample;
42 import org.apache.hadoop.hbase.testclassification.MediumTests;
43 import org.apache.hadoop.hbase.testclassification.MiscTests;
44 import org.apache.hadoop.hbase.util.Bytes;
45 import org.apache.spark.api.java.*;
46 import org.apache.spark.api.java.function.Function;
47 import org.junit.*;
48 import org.junit.experimental.categories.Category;
49
50 import scala.Tuple2;
51
52 import com.google.common.io.Files;
53
54 @Category({MiscTests.class, MediumTests.class})
55 public class TestJavaHBaseContext implements Serializable {
56 private transient JavaSparkContext jsc;
57 HBaseTestingUtility htu;
58 protected static final Log LOG = LogFactory.getLog(TestJavaHBaseContext.class);
59
60
61 byte[] tableName = Bytes.toBytes("t1");
62 byte[] columnFamily = Bytes.toBytes("c");
63 String columnFamilyStr = Bytes.toString(columnFamily);
64
65 @Before
66 public void setUp() {
67 jsc = new JavaSparkContext("local", "JavaHBaseContextSuite");
68
69 File tempDir = Files.createTempDir();
70 tempDir.deleteOnExit();
71
72 htu = HBaseTestingUtility.createLocalHTU();
73 try {
74 LOG.info("cleaning up test dir");
75
76 htu.cleanupTestDir();
77
78 LOG.info("starting minicluster");
79
80 htu.startMiniZKCluster();
81 htu.startMiniHBaseCluster(1, 1);
82
83 LOG.info(" - minicluster started");
84
85 try {
86 htu.deleteTable(TableName.valueOf(tableName));
87 } catch (Exception e) {
88 LOG.info(" - no table " + Bytes.toString(tableName) + " found");
89 }
90
91 LOG.info(" - creating table " + Bytes.toString(tableName));
92 htu.createTable(TableName.valueOf(tableName),
93 columnFamily);
94 LOG.info(" - created table");
95 } catch (Exception e1) {
96 throw new RuntimeException(e1);
97 }
98 }
99
100 @After
101 public void tearDown() {
102 try {
103 htu.deleteTable(TableName.valueOf(tableName));
104 LOG.info("shuting down minicluster");
105 htu.shutdownMiniHBaseCluster();
106 htu.shutdownMiniZKCluster();
107 LOG.info(" - minicluster shut down");
108 htu.cleanupTestDir();
109 } catch (Exception e) {
110 throw new RuntimeException(e);
111 }
112 jsc.stop();
113 jsc = null;
114 }
115
116 @Test
117 public void testBulkPut() throws IOException {
118
119 List<String> list = new ArrayList<>();
120 list.add("1," + columnFamilyStr + ",a,1");
121 list.add("2," + columnFamilyStr + ",a,2");
122 list.add("3," + columnFamilyStr + ",a,3");
123 list.add("4," + columnFamilyStr + ",a,4");
124 list.add("5," + columnFamilyStr + ",a,5");
125
126 JavaRDD<String> rdd = jsc.parallelize(list);
127
128 Configuration conf = htu.getConfiguration();
129
130 JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
131
132 Connection conn = ConnectionFactory.createConnection(conf);
133 Table table = conn.getTable(TableName.valueOf(tableName));
134
135 try {
136 List<Delete> deletes = new ArrayList<>();
137 for (int i = 1; i < 6; i++) {
138 deletes.add(new Delete(Bytes.toBytes(Integer.toString(i))));
139 }
140 table.delete(deletes);
141 } finally {
142 table.close();
143 }
144
145 hbaseContext.bulkPut(rdd,
146 TableName.valueOf(tableName),
147 new PutFunction());
148
149 table = conn.getTable(TableName.valueOf(tableName));
150
151 try {
152 Result result1 = table.get(new Get(Bytes.toBytes("1")));
153 Assert.assertNotNull("Row 1 should had been deleted", result1.getRow());
154
155 Result result2 = table.get(new Get(Bytes.toBytes("2")));
156 Assert.assertNotNull("Row 2 should had been deleted", result2.getRow());
157
158 Result result3 = table.get(new Get(Bytes.toBytes("3")));
159 Assert.assertNotNull("Row 3 should had been deleted", result3.getRow());
160
161 Result result4 = table.get(new Get(Bytes.toBytes("4")));
162 Assert.assertNotNull("Row 4 should had been deleted", result4.getRow());
163
164 Result result5 = table.get(new Get(Bytes.toBytes("5")));
165 Assert.assertNotNull("Row 5 should had been deleted", result5.getRow());
166 } finally {
167 table.close();
168 conn.close();
169 }
170 }
171
172 public static class PutFunction implements Function<String, Put> {
173
174 private static final long serialVersionUID = 1L;
175
176 public Put call(String v) throws Exception {
177 String[] cells = v.split(",");
178 Put put = new Put(Bytes.toBytes(cells[0]));
179
180 put.addColumn(Bytes.toBytes(cells[1]), Bytes.toBytes(cells[2]),
181 Bytes.toBytes(cells[3]));
182 return put;
183 }
184 }
185
186 @Test
187 public void testBulkDelete() throws IOException {
188 List<byte[]> list = new ArrayList<>();
189 list.add(Bytes.toBytes("1"));
190 list.add(Bytes.toBytes("2"));
191 list.add(Bytes.toBytes("3"));
192
193 JavaRDD<byte[]> rdd = jsc.parallelize(list);
194
195 Configuration conf = htu.getConfiguration();
196
197 populateTableWithMockData(conf, TableName.valueOf(tableName));
198
199 JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
200
201 hbaseContext.bulkDelete(rdd, TableName.valueOf(tableName),
202 new JavaHBaseBulkDeleteExample.DeleteFunction(), 2);
203
204
205
206 try (
207 Connection conn = ConnectionFactory.createConnection(conf);
208 Table table = conn.getTable(TableName.valueOf(tableName))
209 ){
210 Result result1 = table.get(new Get(Bytes.toBytes("1")));
211 Assert.assertNull("Row 1 should had been deleted", result1.getRow());
212
213 Result result2 = table.get(new Get(Bytes.toBytes("2")));
214 Assert.assertNull("Row 2 should had been deleted", result2.getRow());
215
216 Result result3 = table.get(new Get(Bytes.toBytes("3")));
217 Assert.assertNull("Row 3 should had been deleted", result3.getRow());
218
219 Result result4 = table.get(new Get(Bytes.toBytes("4")));
220 Assert.assertNotNull("Row 4 should had been deleted", result4.getRow());
221
222 Result result5 = table.get(new Get(Bytes.toBytes("5")));
223 Assert.assertNotNull("Row 5 should had been deleted", result5.getRow());
224 }
225 }
226
227 @Test
228 public void testDistributedScan() throws IOException {
229 Configuration conf = htu.getConfiguration();
230
231 populateTableWithMockData(conf, TableName.valueOf(tableName));
232
233 JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
234
235 Scan scan = new Scan();
236 scan.setCaching(100);
237
238 JavaRDD<String> javaRdd =
239 hbaseContext.hbaseRDD(TableName.valueOf(tableName), scan)
240 .map(new ScanConvertFunction());
241
242 List<String> results = javaRdd.collect();
243
244 Assert.assertEquals(results.size(), 5);
245 }
246
247 private static class ScanConvertFunction implements
248 Function<Tuple2<ImmutableBytesWritable, Result>, String> {
249 @Override
250 public String call(Tuple2<ImmutableBytesWritable, Result> v1) throws Exception {
251 return Bytes.toString(v1._1().copyBytes());
252 }
253 }
254
255 @Test
256 public void testBulkGet() throws IOException {
257 List<byte[]> list = new ArrayList<>();
258 list.add(Bytes.toBytes("1"));
259 list.add(Bytes.toBytes("2"));
260 list.add(Bytes.toBytes("3"));
261 list.add(Bytes.toBytes("4"));
262 list.add(Bytes.toBytes("5"));
263
264 JavaRDD<byte[]> rdd = jsc.parallelize(list);
265
266 Configuration conf = htu.getConfiguration();
267
268 populateTableWithMockData(conf, TableName.valueOf(tableName));
269
270 JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);
271
272 final JavaRDD<String> stringJavaRDD =
273 hbaseContext.bulkGet(TableName.valueOf(tableName), 2, rdd,
274 new GetFunction(),
275 new ResultFunction());
276
277 Assert.assertEquals(stringJavaRDD.count(), 5);
278 }
279
280 public static class GetFunction implements Function<byte[], Get> {
281
282 private static final long serialVersionUID = 1L;
283
284 public Get call(byte[] v) throws Exception {
285 return new Get(v);
286 }
287 }
288
289 public static class ResultFunction implements Function<Result, String> {
290
291 private static final long serialVersionUID = 1L;
292
293 public String call(Result result) throws Exception {
294 Iterator<Cell> it = result.listCells().iterator();
295 StringBuilder b = new StringBuilder();
296
297 b.append(Bytes.toString(result.getRow())).append(":");
298
299 while (it.hasNext()) {
300 Cell cell = it.next();
301 String q = Bytes.toString(CellUtil.cloneQualifier(cell));
302 if ("counter".equals(q)) {
303 b.append("(")
304 .append(q)
305 .append(",")
306 .append(Bytes.toLong(CellUtil.cloneValue(cell)))
307 .append(")");
308 } else {
309 b.append("(")
310 .append(q)
311 .append(",")
312 .append(Bytes.toString(CellUtil.cloneValue(cell)))
313 .append(")");
314 }
315 }
316 return b.toString();
317 }
318 }
319
320 private void populateTableWithMockData(Configuration conf, TableName tableName)
321 throws IOException {
322 try (
323 Connection conn = ConnectionFactory.createConnection(conf);
324 Table table = conn.getTable(tableName)) {
325
326 List<Put> puts = new ArrayList<>();
327
328 for (int i = 1; i < 6; i++) {
329 Put put = new Put(Bytes.toBytes(Integer.toString(i)));
330 put.addColumn(columnFamily, columnFamily, columnFamily);
331 puts.add(put);
332 }
333 table.put(puts);
334 }
335 }
336
337 }