HBase Java編程示例

HBase Java編程示例

1. package elementary;
   
2. 
   
3. import java.io.IOException;
   
4. import java.text.SimpleDateFormat;
   
5. import java.util.ArrayList;
   
6. import java.util.Date;
   
7. import java.util.List;
   
8. import java.util.concurrent.atomic.AtomicInteger;
   
9. import java.util.concurrent.ExecutorService;
   
10. import java.util.concurrent.Executors;
    
11. import java.util.concurrent.TimeUnit;
    
12. 
    
13. import org.apache.hadoop.conf.Configuration;
    
14. import org.apache.hadoop.hbase.Cell;
    
15. import org.apache.hadoop.hbase.HBaseConfiguration;
    
16. import org.apache.hadoop.hbase.HColumnDescriptor;
    
17. import org.apache.hadoop.hbase.HTableDescriptor;
    
18. import org.apache.hadoop.hbase.MasterNotRunningException;
    
19. import org.apache.hadoop.hbase.TableName;
    
20. import org.apache.hadoop.hbase.ZooKeeperConnectionException;
    
21. import org.apache.hadoop.hbase.client.Delete;
    
22. import org.apache.hadoop.hbase.client.Get;
    
23. import org.apache.hadoop.hbase.client.Admin;
    
24. import org.apache.hadoop.hbase.client.BufferedMutator;
    
25. import org.apache.hadoop.hbase.client.BufferedMutatorParams;
    
26. import org.apache.hadoop.hbase.client.Connection;
    
27. import org.apache.hadoop.hbase.client.ConnectionFactory;
    
28. import org.apache.hadoop.hbase.client.Table;
    
29. import org.apache.hadoop.hbase.client.Put;
    
30. import org.apache.hadoop.hbase.client.Result;
    
31. import org.apache.hadoop.hbase.client.ResultScanner;
    
32. import org.apache.hadoop.hbase.client.Scan;
    
33. import org.apache.hadoop.hbase.util.Bytes;
    
34. import org.apache.hadoop.util.ThreadUtil;
    
35. 
    
36. public class HelloWorld {
    
37. private static Configuration conf = null;
    
38. private static Connection conn = null;
    
39. private static Admin admin = null;
    
40. public static AtomicInteger count = new AtomicInteger();
    
41. 
    
42. /**
    
43. * 初始化配置
    
44. */
    
45. static {
    
46. conf = HBaseConfiguration.create();
    
47. //如果沒有配置文件，一定要記得手動宣告
    
48. 
    
49. conf.set("hbase.zookeeper.quorum", "10.148.137.143");
    
50. conf.set("hbase.zookeeper.property.clientPort", "2181");
    
51. }
    
52. 
    
53. static {
    
54. try {
    
55. conn = ConnectionFactory.createConnection();
    
56. admin = conn.getAdmin();
    
57. } catch (IOException e) {
    
58. e.printStackTrace();
    
59. }
    
60. }
    
61. 
    
62. static public class MyThread extends Thread
    
63. {
    
64. int _start;
    
65. String _tablename;
    
66. Connection conn;
    
67. //BufferedMutator table;
    
68. Table table;
    
69. 
    
70. public MyThread(int start, String tablename) {
    
71. _start = start;
    
72. _tablename = tablename;
    
73. }
    
74. 
    
75. public void run() {
    
76. String tablename = _tablename;
    
77. Thread current = Thread.currentThread();
    
78. long thread_id = current.getId();
    
79. System.out.printf("thread[%d] run\n", thread_id);
    
80. 
    
81. try {
    
82. conn = ConnectionFactory.createConnection();
    
83. //BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tablename));
    
84. //params.writeBufferSize(1024 * 4);
    
85. //table = conn.getBufferedMutator(params);
    
86. table = conn.getTable(TableName.valueOf(tablename));
    
87. 
    
88. for (int j=_start; j<100; ++j) {
    
89. for (int i=0; i<10000000; ++i) {
    
90. // zkb_0_0
    
91. String zkb = "zkb_" + String.valueOf(_start) + "_" + String.valueOf(i);
    
92. Put put = new Put(Bytes.toBytes(zkb));
    
93. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field1"),Bytes.toBytes(String.valueOf(i+0)));
    
94. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field2"),Bytes.toBytes(String.valueOf(i+1)));
    
95. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field3"),Bytes.toBytes(String.valueOf(i+2)));
    
96. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field4"),Bytes.toBytes(String.valueOf(i+3)));
    
97. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field5"),Bytes.toBytes(String.valueOf(i+4)));
    
98. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field6"),Bytes.toBytes(String.valueOf(i+5)));
    
99. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field7"),Bytes.toBytes(String.valueOf(i+6)));
    
100. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field8"),Bytes.toBytes(String.valueOf(i+7)));
     
101. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field9"),Bytes.toBytes(String.valueOf(i+8)));
     
102. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field10"),Bytes.toBytes(String.valueOf(i+9)));
     
103. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field11"),Bytes.toBytes(String.valueOf(i+10)));
     
104. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field12"),Bytes.toBytes(String.valueOf(i+11)));
     
105. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field13"),Bytes.toBytes(String.valueOf(i+12)));
     
106. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field14"),Bytes.toBytes(String.valueOf(i+13)));
     
107. put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field15"),Bytes.toBytes(String.valueOf(i+14)));
     
108. //table.mutate(put);
     
109. table.put(put);
     
110. 
     
111. int m = HelloWorld.count.incrementAndGet();
     
112. if (m % 10000 == 0) {
     
113. Date dt = new Date();
     
114. SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss aa");
     
115. String now = sdf.format(dt);
     
116. System.out.printf("[%s] thread[%d] m=%d, j=%d, i=%d\n", now, thread_id, m, j, i);
     
117. }
     
118. }
     
119. }
     
120. 
     
121. System.out.printf("thread[%d] over\n", thread_id);
     
122. }
     
123. catch (Exception e) {
     
124. e.printStackTrace();
     
125. }
     
126. }
     
127. }
     
128. 
     
129. /**
     
130. * 建立表格
     
131. * @param tablename
     
132. * @param cfs
     
133. */
     
134. public static void createTable(String tablename, String[] cfs){
     
135. try {
     
136. if (admin.tableExists(TableName.valueOf(tablename))) {
     
137. System.out.println("table already exists!");
     
138. } else {
     
139. HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tablename));
     
140. for (int i = 0; i < cfs.length; i++) {
     
141. HColumnDescriptor desc = new HColumnDescriptor(cfs[i]);
     
142. desc.setMaxVersions(3650);
     
143. tableDesc.addFamily(desc);
     
144. }
     
145. 
     
146. byte[][] splitKeys = new byte[][] {
     
147. Bytes.toBytes("zkb_0_0"),
     
148. Bytes.toBytes("zkb_10_0"),
     
149. Bytes.toBytes("zkb_20_0"),
     
150. Bytes.toBytes("zkb_30_0"),
     
151. Bytes.toBytes("zkb_40_0"),
     
152. Bytes.toBytes("zkb_50_0"),
     
153. Bytes.toBytes("zkb_60_0"),
     
154. Bytes.toBytes("zkb_70_0"),
     
155. Bytes.toBytes("zkb_80_0"),
     
156. Bytes.toBytes("zkb_90_0"),
     
157. Bytes.toBytes("zkb_100_0")
     
158. };
     
159. admin.createTable(tableDesc, splitKeys);
     
160. admin.close();
     
161. System.out.println("create table " + tablename + " ok.");
     
162. }
     
163. } catch (MasterNotRunningException e) {
     
164. e.printStackTrace();
     
165. } catch (ZooKeeperConnectionException e) {
     
166. e.printStackTrace();
     
167. } catch (IOException e) {
     
168. e.printStackTrace();
     
169. }
     
170. }
     
171. 
     
172. /**
     
173. * 刪除表格
     
174. * @param tablename
     
175. */
     
176. public static void deleteTable(String tablename){
     
177. try {
     
178. //Connection conn = ConnectionFactory.createConnection();
     
179. //Admin admin = conn.getAdmin();
     
180. admin.disableTable(TableName.valueOf(tablename));
     
181. admin.deleteTable(TableName.valueOf(tablename));
     
182. System.out.println("delete table " + tablename + " ok.");
     
183. } catch (IOException e) {
     
184. e.printStackTrace();
     
185. }
     
186. }
     
187. 
     
188. /**
     
189. * 刪除一筆資料
     
190. * @param tableName
     
191. * @param rowKey
     
192. */
     
193. public static void delRecord (String tableName, String rowKey){
     
194. try {
     
195. Table table = conn.getTable(TableName.valueOf(tableName));
     
196. 
     
197. List<Delete> list = new ArrayList<Delete>();
     
198. Delete del = new Delete(rowKey.getBytes());
     
199. list.add(del);
     
200. table.delete(list);
     
201. System.out.println("del recored " + rowKey + " ok.");
     
202. } catch (IOException e) {
     
203. e.printStackTrace();
     
204. }
     
205. }
     
206. 
     
207. /**
     
208. * 取得一筆資料
     
209. * @param tableName
     
210. * @param rowKey
     
211. */
     
212. public static void getOneRecord (String tableName, String rowKey){
     
213. try {
     
214. Table table = conn.getTable(TableName.valueOf(tableName));
     
215. 
     
216. Get get = new Get(rowKey.getBytes());
     
217. Result rs = table.get(get);
     
218. List<Cell> list = rs.listCells();
     
219. for(Cell cell:list){
     
220. System.out.print(new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()) + " " );
     
221. System.out.print(new String(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()) + ":" );
     
222. System.out.print(new String(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()) + " " );
     
223. System.out.print(cell.getTimestamp() + " " );
     
224. System.out.print(new String(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()) + " " );
     
225. System.out.println("");
     
226. }
     
227. } catch (IOException e) {
     
228. e.printStackTrace();
     
229. }
     
230. }
     
231. 
     
232. /**
     
233. * 取得所有資料
     
234. * @param tableName
     
235. */
     
236. public static void getAllRecord (String tableName) {
     
237. try{
     
238. //Connection conn = ConnectionFactory.createConnection();
     
239. Table table = conn.getTable(TableName.valueOf(tableName));
     
240. 
     
241. Scan scan = new Scan();
     
242. ResultScanner resultscanner = table.getScanner(scan);
     
243. for(Result rs:resultscanner){
     
244. List<Cell> list = rs.listCells();
     
245. for(Cell cell:list){
     
246. System.out.print(new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()) + " " );
     
247. System.out.print(new String(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()) + ":" );
     
248. System.out.print(new String(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()) + " " );
     
249. System.out.print(cell.getTimestamp() + " " );
     
250. System.out.print(new String(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()) + " " );
     
251. System.out.println("");
     
252. }
     
253. }
     
254. } catch (IOException e){
     
255. e.printStackTrace();
     
256. }
     
257. }
     
258. 
     
259. /**
     
260. * 取得Family清單
     
261. * @param tableName
     
262. * @return
     
263. */
     
264. public static ArrayList<String> getAllFamilyName(String tableName) {
     
265. ArrayList<String> familyname_list = new ArrayList<String>();
     
266. try{
     
267. //Connection conn = ConnectionFactory.createConnection();
     
268. Table table = conn.getTable(TableName.valueOf(tableName));
     
269. 
     
270. HTableDescriptor htabledescriptor = table.getTableDescriptor();
     
271. HColumnDescriptor[] hdlist = htabledescriptor.getColumnFamilies();
     
272. for(int i=0;i<hdlist.length;i++){
     
273. HColumnDescriptor hd = hdlist[i];
     
274. familyname_list.add(hd.getNameAsString());
     
275. }
     
276. } catch (IOException e){
     
277. e.printStackTrace();
     
278. }
     
279. return familyname_list;
     
280. }
     
281. 
     
282. // java -cp HelloWorld.jar:`ls lib/*.jar|awk '{printf("%s:", $0)}'` elementary.HelloWorld 5
     
283. public static void main(String[] args) {
     
284. System.out.println("HelloWorldX");
     
285. if (args.length > 0)
     
286. System.out.println(args[0]);
     
287. 
     
288. int start = 0;
     
289. if (args.length > 1)
     
290. start = Integer.valueOf(args[1]);
     
291. if (start < 0)
     
292. start = 0;
     
293. 
     
294. int num_threads = 16;
     
295. if (args.length > 2)
     
296. num_threads = Integer.valueOf(args[2]);
     
297. 
     
298. try {
     
299. String tablename = "scores";
     
300. String[] familys = {"grade", "course"};
     
301. HelloWorld.createTable(tablename, familys);
     
302. 
     
303. //ExecutorService thread_pool = Executors.newSingleThreadExecutor();
     
304. ExecutorService thread_pool = Executors.newFixedThreadPool(num_threads);
     
305. Thread[] pool = new HelloWorld.MyThread[80];
     
306. for (int i=0; i<pool.length; ++i) {
     
307. pool[i] = new HelloWorld.MyThread(i, tablename);
     
308. thread_pool.execute(pool[i]);
     
309. }
     
310. 
     
311. thread_pool.shutdown();
     
312. System.out.println("over");
     
313. }
     
314. catch (Exception e) {
     
315. e.printStackTrace();
     
316. }
     
317. }
     
318. 
     
319. }