新浪微博數據解析與java操做Hbase實例

以前發過一篇開發新浪微博的文章，對於你們比較感興趣的內容之一即是如何解析新浪微博的JSON。

其實一開始的時候，也遇過一些挫折，好比直接用JsonArray和JsonObject去解析JSON內容的話，是解析不了的。

由於JSON的格式比較固定，像新浪微博返回的JSON內容則是多了一箇中括號及statues標籤，以下：

{ "statuses": [ {
            "created_at": "Tue May 31 17:46:55 +0800 2011",
            "id": 11488058246,
            "text": "求關注。",
            "source": "<a href="http://weibo.com" rel="nofollow">新浪微博</a>",
            "favorited": false,
            "truncated": false,
            "in_reply_to_status_id": "",
            "in_reply_to_user_id": "",
            "in_reply_to_screen_name": "",
            "geo": null,
            "mid": "5612814510546515491",
            "reposts_count": 8,
            "comments_count": 9,
            "annotations": [],
            "user": {
                "id": 1404376560,
                "screen_name": "zaku",
                "name": "zaku",
                "province": "11",
                "city": "5",
                "location": "北京 朝陽區",
                "description": "人生五十年，乃如夢如幻；有生斯有死，壯士復何憾。",
                "url": "http://blog.sina.com.cn/zaku",
                "profile_image_url": "http://tp1.sinaimg.cn/1404376560/50/0/1",
                "domain": "zaku",
                "gender": "m",
                "followers_count": 1204,
                "friends_count": 447,
                "statuses_count": 2908,
                "favourites_count": 0,
                "created_at": "Fri Aug 28 00:00:00 +0800 2009",
                "following": false,
                "allow_all_act_msg": false,
                "remark": "",
                "geo_enabled": true,
                "verified": false,
                "allow_all_comment": true,
                "avatar_large": "http://tp1.sinaimg.cn/1404376560/180/0/1",
                "verified_reason": "",
                "follow_me": false,
                "online_status": 0,
                "bi_followers_count": 215
            }
        },
        ...  ], "previous_cursor": 0, // 暫未支持 "next_cursor": 11488013766, // 暫未支持 "total_number": 81655 }

個人目標就是先去掉以上紅色字體的內容，這個比較容易實現。

[java]   view plain copy

1. //將獲取到的json內容變成能夠正常解析的文本  
2.     public static String parseJsonText(String jtext){  
3.           
4.         char leftSymbol='[';  
5.         char rightSymbol=']';  
6.         String result="";  
7.         for(int i=0;i<jtext.length();i++){  
8.               
9.             if(leftSymbol==jtext.charAt(i)){  
10.                 result=jtext.substring(i,jtext.length());  
11.                 break;  
12.             }  
13.         }  
14.           
15.         for(int i=result.length();i>0;i--){  
16.               
17.             if(rightSymbol==result.charAt(i-1)){  
18.                 result=result.substring(0, i);  
19.                 break;  
20.             }  
21.         }  
22.         return result;  
23.           
24.     }  

處理完就能夠直接用JsonArray和JsonObject來解析相關的內容了。

第二步就是實現一個List<HashMap<String,String>>,用於填充ListView。

下面的方法返回一個List<HashMap<String,String>>,並在其中解析JSON內容。

[java]   view plain copy

1. private List<HashMap<String, String>> listview(String rlt) {  
2.   
3.         String result = ParseJson.parseJsonText(rlt);  
4.         try {  
5.             JSONArray jsonArrays = new JSONArray(result);  
6.             for (int i = 0; i < jsonArrays.length(); i++) {  
7.                 JSONObject jsonObject = jsonArrays.getJSONObject(i);  
8.   
9.                 HashMap<String, String> map = new HashMap<String, String>();  
10.                 // 得到微博的ID  
11.                 String ID = jsonObject.getString("id");  
12.                 contentId.add(ID);  
13.                 // 得到微博內容  
14.                 String text = jsonObject.getString("text");  
15.                 // 得到微博來源  
16.                 String source = jsonObject.getString("source");  
17.                 // 得到微博的發表時間  
18.                 String created_at = jsonObject.getString("created_at");  
19.                 // 生成user的可用JSON對象  
20.                 JSONObject userJson = jsonObject.getJSONObject("user");  
21.                 // 得到發表微博的用戶名  
22.                 String username = userJson.getString("name");  
23.   
24.                 map.put("text", text);  
25.                 map.put("name", username);  
26.                 map.put("source", "來自" + "【" + source + "】");  
27.                 map.put("created_at", ParseJson.getFormatTime(created_at));  
28.   
29.                 if (jsonObject.has("retweeted_status")) {  
30.                     JSONObject retweetedJson = jsonObject  
31.                             .getJSONObject("retweeted_status");  
32.                     // 得到轉發微博的內容  
33.                     String retweetedText = retweetedJson.getString("text");  
34.                     map.put("retweetedText", retweetedText);  
35.                     JSONObject retWeetedUserNameJson = retweetedJson  
36.                             .getJSONObject("user");  
37.                     // 得到轉發微博內容的用戶名  
38.                     String retweetedContent = retWeetedUserNameJson  
39.                             .getString("name");  
40.                     map.put("retweetedContent", retweetedContent + ":"  
41.                             + retweetedText);  
42.                 }  
43.                 listArrays.add(map);  
44.             }  
45.   
46.         } catch (JSONException e) {  
47.             // TODO Auto-generated catch block  
48.             System.out.println(e.getMessage());  
49.         }  
50.   
51.         return listArrays;  
52.   
53.     }  

經過上面兩步就能夠實現基本的微博JSON解析了。

Java操做Hbase進行建表、刪表以及對數據進行增刪改查，條件查詢

博客分類：  
• SQL/NOSQL

 

一、搭建環境

  新建JAVA項目，添加的包有:

   有關Hadoop的hadoop-core-0.20.204.0.jar

   有關Hbase的hbase-0.90.4.jar、hbase-0.90.4-tests.jar以及Hbase資源包中lib目錄下的全部jar包

 

二、主要程序

 

Java代碼    

1. package com.wujintao.hbase.test;  
2.   
3. import java.io.IOException;  
4. import java.util.ArrayList;  
5. import java.util.List;  
6.   
7. import org.apache.hadoop.conf.Configuration;  
8. import org.apache.hadoop.hbase.HBaseConfiguration;  
9. import org.apache.hadoop.hbase.HColumnDescriptor;  
10. import org.apache.hadoop.hbase.HTableDescriptor;  
11. import org.apache.hadoop.hbase.KeyValue;  
12. import org.apache.hadoop.hbase.MasterNotRunningException;  
13. import org.apache.hadoop.hbase.ZooKeeperConnectionException;  
14. import org.apache.hadoop.hbase.client.Delete;  
15. import org.apache.hadoop.hbase.client.Get;  
16. import org.apache.hadoop.hbase.client.HBaseAdmin;  
17. import org.apache.hadoop.hbase.client.HTable;  
18. import org.apache.hadoop.hbase.client.HTablePool;  
19. import org.apache.hadoop.hbase.client.Put;  
20. import org.apache.hadoop.hbase.client.Result;  
21. import org.apache.hadoop.hbase.client.ResultScanner;  
22. import org.apache.hadoop.hbase.client.Scan;  
23. import org.apache.hadoop.hbase.filter.Filter;  
24. import org.apache.hadoop.hbase.filter.FilterList;  
25. import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;  
26. import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;  
27. import org.apache.hadoop.hbase.util.Bytes;  
28.   
29. public class JinTaoTest {  
30.   
31.     public static Configuration configuration;  
32.     static {  
33.         configuration = HBaseConfiguration.create();  
34.         configuration.set("hbase.zookeeper.property.clientPort", "2181");  
35.         configuration.set("hbase.zookeeper.quorum", "192.168.1.100");  
36.         configuration.set("hbase.master", "192.168.1.100:600000");  
37.     }  
38.   
39.     public static void main(String[] args) {  
40.         // createTable("wujintao");  
41.         // insertData("wujintao");  
42.         // QueryAll("wujintao");  
43.         // QueryByCondition1("wujintao");  
44.         // QueryByCondition2("wujintao");  
45.         //QueryByCondition3("wujintao");  
46.         //deleteRow("wujintao","abcdef");  
47.         deleteByCondition("wujintao","abcdef");  
48.     }  
49.   
50.     /** 
51.      * 建立表 
52.      * @param tableName 
53.      */  
54.     public static void createTable(String tableName) {  
55.         System.out.println("start create table ......");  
56.         try {  
57.             HBaseAdmin hBaseAdmin = new HBaseAdmin(configuration);  
58.             if (hBaseAdmin.tableExists(tableName)) {// 若是存在要建立的表，那麼先刪除，再建立  
59.                 hBaseAdmin.disableTable(tableName);  
60.                 hBaseAdmin.deleteTable(tableName);  
61.                 System.out.println(tableName + " is exist,detele....");  
62.             }  
63.             HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);  
64.             tableDescriptor.addFamily(new HColumnDescriptor("column1"));  
65.             tableDescriptor.addFamily(new HColumnDescriptor("column2"));  
66.             tableDescriptor.addFamily(new HColumnDescriptor("column3"));  
67.             hBaseAdmin.createTable(tableDescriptor);  
68.         } catch (MasterNotRunningException e) {  
69.             e.printStackTrace();  
70.         } catch (ZooKeeperConnectionException e) {  
71.             e.printStackTrace();  
72.         } catch (IOException e) {  
73.             e.printStackTrace();  
74.         }  
75.         System.out.println("end create table ......");  
76.     }  
77.   
78.     /** 
79.      * 插入數據 
80.      * @param tableName 
81.      */  
82.     public static void insertData(String tableName) {  
83.         System.out.println("start insert data ......");  
84.         HTablePool pool = new HTablePool(configuration, 1000);  
85.         HTable table = (HTable) pool.getTable(tableName);  
86.         Put put = new Put("112233bbbcccc".getBytes());// 一個PUT表明一行數據，再NEW一個PUT表示第二行數據,每行一個惟一的ROWKEY，此處rowkey爲put構造方法中傳入的值  
87.         put.add("column1".getBytes(), null, "aaa".getBytes());// 本行數據的第一列  
88.         put.add("column2".getBytes(), null, "bbb".getBytes());// 本行數據的第三列  
89.         put.add("column3".getBytes(), null, "ccc".getBytes());// 本行數據的第三列  
90.         try {  
91.             table.put(put);  
92.         } catch (IOException e) {  
93.             e.printStackTrace();  
94.         }  
95.         System.out.println("end insert data ......");  
96.     }  
97.   
98.     /** 
99.      * 刪除一張表 
100.      * @param tableName 
101.      */  
102.     public static void dropTable(String tableName) {  
103.         try {  
104.             HBaseAdmin admin = new HBaseAdmin(configuration);  
105.             admin.disableTable(tableName);  
106.             admin.deleteTable(tableName);  
107.         } catch (MasterNotRunningException e) {  
108.             e.printStackTrace();  
109.         } catch (ZooKeeperConnectionException e) {  
110.             e.printStackTrace();  
111.         } catch (IOException e) {  
112.             e.printStackTrace();  
113.         }  
114.   
115.     }  
116.     /** 
117.      * 根據 rowkey刪除一條記錄 
118.      * @param tablename 
119.      * @param rowkey 
120.      */  
121.      public static void deleteRow(String tablename, String rowkey)  {  
122.         try {  
123.             HTable table = new HTable(configuration, tablename);  
124.             List list = new ArrayList();  
125.             Delete d1 = new Delete(rowkey.getBytes());  
126.             list.add(d1);  
127.               
128.             table.delete(list);  
129.             System.out.println("刪除行成功!");  
130.               
131.         } catch (IOException e) {  
132.             e.printStackTrace();  
133.         }  
134.           
135.   
136.     }  
137.   
138.      /** 
139.       * 組合條件刪除 
140.       * @param tablename 
141.       * @param rowkey 
142.       */  
143.      public static void deleteByCondition(String tablename, String rowkey)  {  
144.             //目前尚未發現有效的API可以實現 根據非rowkey的條件刪除 這個功能能，還有清空表所有數據的API操做  
145.   
146.     }  
147.   
148.   
149.     /** 
150.      * 查詢全部數據 
151.      * @param tableName 
152.      */  
153.     public static void QueryAll(String tableName) {  
154.         HTablePool pool = new HTablePool(configuration, 1000);  
155.         HTable table = (HTable) pool.getTable(tableName);  
156.         try {  
157.             ResultScanner rs = table.getScanner(new Scan());  
158.             for (Result r : rs) {  
159.                 System.out.println("得到到rowkey:" + new String(r.getRow()));  
160.                 for (KeyValue keyValue : r.raw()) {  
161.                     System.out.println("列：" + new String(keyValue.getFamily())  
162.                             + "====值:" + new String(keyValue.getValue()));  
163.                 }  
164.             }  
165.         } catch (IOException e) {  
166.             e.printStackTrace();  
167.         }  
168.     }  
169.   
170.     /** 
171.      * 單條件查詢,根據rowkey查詢惟一一條記錄 
172.      * @param tableName 
173.      */  
174.     public static void QueryByCondition1(String tableName) {  
175.   
176.         HTablePool pool = new HTablePool(configuration, 1000);  
177.         HTable table = (HTable) pool.getTable(tableName);  
178.         try {  
179.             Get scan = new Get("abcdef".getBytes());// 根據rowkey查詢  
180.             Result r = table.get(scan);  
181.             System.out.println("得到到rowkey:" + new String(r.getRow()));  
182.             for (KeyValue keyValue : r.raw()) {  
183.                 System.out.println("列：" + new String(keyValue.getFamily())  
184.                         + "====值:" + new String(keyValue.getValue()));  
185.             }  
186.         } catch (IOException e) {  
187.             e.printStackTrace();  
188.         }  
189.     }  
190.   
191.     /** 
192.      * 單條件按查詢，查詢多條記錄 
193.      * @param tableName 
194.      */  
195.     public static void QueryByCondition2(String tableName) {  
196.   
197.         try {  
198.             HTablePool pool = new HTablePool(configuration, 1000);  
199.             HTable table = (HTable) pool.getTable(tableName);  
200.             Filter filter = new SingleColumnValueFilter(Bytes  
201.                     .toBytes("column1"), null, CompareOp.EQUAL, Bytes  
202.                     .toBytes("aaa")); // 當列column1的值爲aaa時進行查詢  
203.             Scan s = new Scan();  
204.             s.setFilter(filter);  
205.             ResultScanner rs = table.getScanner(s);  
206.             for (Result r : rs) {  
207.                 System.out.println("得到到rowkey:" + new String(r.getRow()));  
208.                 for (KeyValue keyValue : r.raw()) {  
209.                     System.out.println("列：" + new String(keyValue.getFamily())  
210.                             + "====值:" + new String(keyValue.getValue()));  
211.                 }  
212.             }  
213.         } catch (Exception e) {  
214.             e.printStackTrace();  
215.         }  
216.   
217.     }  
218.   
219.     /** 
220.      * 組合條件查詢 
221.      * @param tableName 
222.      */  
223.     public static void QueryByCondition3(String tableName) {  
224.   
225.         try {  
226.             HTablePool pool = new HTablePool(configuration, 1000);  
227.             HTable table = (HTable) pool.getTable(tableName);  
228.   
229.             List<Filter> filters = new ArrayList<Filter>();  
230.   
231.             Filter filter1 = new SingleColumnValueFilter(Bytes  
232.                     .toBytes("column1"), null, CompareOp.EQUAL, Bytes  
233.                     .toBytes("aaa"));  
234.             filters.add(filter1);  
235.   
236.             Filter filter2 = new SingleColumnValueFilter(Bytes  
237.                     .toBytes("column2"), null, CompareOp.EQUAL, Bytes  
238.                     .toBytes("bbb"));  
239.             filters.add(filter2);  
240.   
241.             Filter filter3 = new SingleColumnValueFilter(Bytes  
242.                     .toBytes("column3"), null, CompareOp.EQUAL, Bytes  
243.                     .toBytes("ccc"));  
244.             filters.add(filter3);  
245.   
246.             FilterList filterList1 = new FilterList(filters);  
247.   
248.             Scan scan = new Scan();  
249.             scan.setFilter(filterList1);  
250.             ResultScanner rs = table.getScanner(scan);  
251.             for (Result r : rs) {  
252.                 System.out.println("得到到rowkey:" + new String(r.getRow()));  
253.                 for (KeyValue keyValue : r.raw()) {  
254.                     System.out.println("列：" + new String(keyValue.getFamily())  
255.                             + "====值:" + new String(keyValue.getValue()));  
256.                 }  
257.             }  
258.             rs.close();  
259.   
260.         } catch (Exception e) {  
261.             e.printStackTrace();  
262.         }  
263.   
264.     }  
265.   
266. }  

 注意：可能你們沒看到更新數據的操做，其實更新的操做跟添加徹底一致，只不過是添加呢rowkey不存在，更新呢rowkey已經存在，而且timstamp相同的狀況下，還有就是目前好像還沒辦法實現hbase數據的分頁查詢，不知道有沒有人知道怎麼作

 

HBase性能優化建議：

 針對前面的代碼，有不少不足之處，在此我就不修改上面的代碼了，只是提出建議的地方，你們本身加上

   1)配置

  當你調用create方法時將會加載兩個配置文件:hbase-default.xml and hbase-site.xml,利用的是當前的java類路徑， 代碼中configuration設置的這些配置將會覆蓋hbase-default.xml和hbase-site.xml中相同的配置,若是兩個配置文件都存在而且都設置好了相應參上面的屬性下面的屬性便可

 

 2)關於建表

  

public void createTable(HTableDescriptor desc)

 

HTableDescriptor 表明的是表的schema, 提供的方法中比較有用的有

setMaxFileSize，指定最大的region size

setMemStoreFlushSize 指定memstore flush到HDFS上的文件大小

增長family經過 addFamily方法

 

public void addFamily(final HColumnDescriptor family)

 

HColumnDescriptor表明的是column的schema，提供的方法比較經常使用的有

setTimeToLive:指定最大的TTL,單位是ms,過時數據會被自動刪除。

setInMemory:指定是否放在內存中，對小表有用，可用於提升效率。默認關閉

setBloomFilter:指定是否使用BloomFilter,可提升隨機查詢效率。默認關閉

setCompressionType:設定數據壓縮類型。默認無壓縮。

setMaxVersions:指定數據最大保存的版本個數。默認爲3。

 

注意的是，通常咱們不去setInMemory爲true,默認是關閉的

 

3)關於入庫

   官方建議

 table.setAutoFlush(false); //數據入庫以前先設置此項爲false

 table.setflushCommits();//入庫完成後，手動刷入數據

注意：

  在入庫過程當中，put.setWriteToWAL(true/flase);

  關於這一項若是不但願大量數據在存儲過程當中丟失，建議設置爲true,若是僅是在測試演練階段，爲了節省入庫時間建議設置爲false

 

4)關於獲取表實例

HTablePool pool = new HTablePool(configuration, Integer.MAX_VALUE);

HTable table = (HTable) pool.getTable(tableName);

建議用錶鏈接池的方式獲取表，具體池有什麼做用，我想用過數據庫鏈接池的同窗都知道，我就再也不重複

不建議使用new HTable(configuration,tableName);的方式獲取表

 

5)關於查詢

 建議每一個查詢語句都放入try catch語句塊，而且finally中要進行關閉ResultScanner實例以及將不使用的表從新放入到HTablePool中的操做，具體作法以下

Java代碼    

1. public static void QueryAll(String tableName) {  
2.         HTablePool pool = new HTablePool(configuration, Integer.MAX_VALUE);  
3.         HTable table = null;  
4.         ResultScanner rs = null;  
5.         try {  
6.             Scan scan = new Scan();  
7.             table = (HTable) pool.getTable(tableName);  
8.             rs = table.getScanner(scan);  
9.             for (Result r : rs) {  
10.                 System.out.println("得到到rowkey:" + new String(r.getRow()));  
11.                 for (KeyValue keyValue : r.raw()) {  
12.                     System.out.println("列：" + new String(keyValue.getFamily())  
13.                             + "====值:" + new String(keyValue.getValue()));  
14.                 }  
15.             }  
16.         } catch (IOException e) {  
17.             e.printStackTrace();  
18.         }finally{  
19.             rs.close();// 最後還得關閉  
20.             pool.putTable(table); //實際應用過程當中，pool獲取實例的方式應該抽取爲單例模式的，不該在每一個方法都從新獲取一次(單例明白？就是抽取到專門獲取pool的邏輯類中，具體邏輯爲若是pool存在着直接使用，若是不存在則new)  
21.         }  
22.     }  

 

 因此，以上代碼有缺陷的地方，感興趣的同窗能夠針對優化建議做出相應修改

http://blog.csdn.net/ieicihc/article/details/10604129

http://javacrazyer.iteye.com/blog/1186881