Lucene搜索引擎+HDFS+MR完成垂直搜索

介於上一篇的java實現網路爬蟲基礎之上，這一篇的思想是將網路收集的數據保存到HDFS和資料庫（Mysql）中；然後用MR對HDFS的數據進行索引處理，處理成倒排索引；搜索時先用HDFS建立好的索引來搜索對應的數據ID，根據ID從資料庫中提取數據，呈現到網頁上。這是一個完整的集合網路爬蟲、資料庫、 ...

　　介於上一篇的java實現網路爬蟲基礎之上，這一篇的思想是將網路收集的數據保存到HDFS和資料庫（Mysql）中；然後用MR對HDFS的數據進行索引處理，處理成倒排索引；搜索時先用HDFS建立好的索引來搜索對應的數據ID，根據ID從資料庫中提取數據，呈現到網頁上。

　　這是一個完整的集合網路爬蟲、資料庫、HDFS、MapReduce、DAO設計模式、JSP/Servlet的項目，完成了數據收集、數據分析、數據索引並分頁呈現。

　　完整的代碼呈現，希望認真仔細閱讀。

------>

目錄：

1、搜索引擎闡述

2、資料庫表建立

3、使用DAO設計模式進行資料庫操作

【Ⅰ】資料庫連接類DataBaseConnection

【Ⅱ】表單元素的封裝類News

【Ⅲ】編寫DAO介面INewsDAO

【Ⅳ】DAO介面的實現類NewsDAOImp類

【Ⅴ】工廠類DAOFactory類

4、網路爬蟲實現★★　【參考博客《java實現網路爬蟲》和《Heritrix實現網路爬蟲》】

5、MR（MapReduce）對HDFS數據進行索引處理★★

6、實現搜索引擎

【Ⅰ】創建web項目，編寫測試用例，測試是否可以讀取HDFS的數據內容

【Ⅱ】編寫index首頁

【Ⅲ】處理HDFS查詢的操作

【Ⅳ】servlet類搜索結果向頁面傳遞

【Ⅴ】結果呈現，實現分頁

7、總結

------>

1、搜索引擎闡述

搜索引擎的執行流程：

1）通過爬蟲來將數據下載到本地

2）將數據提取出來保存到HDFS和資料庫中（MySQL）

3）通過MR來對HDFS的數據進行索引處理，處理成為倒排索引

4）搜索時先使用HDFS建立好的索引來搜索對應的數據ID，再根據ID來從MySQL資料庫中提取數據的具體信息。

5）可以完成分頁等操作。

倒排索引對應的就是正排索引，正排索引指的就是MySQL資料庫中id的索引。

而倒排索引的目的是可以根據關鍵字查詢出該關鍵字對應的數據id。

這裡就需要用到MySQL資料庫，以及通過Java EE版的Eclipse來完成網站的開發。

為了開發起來更方便，我們這裡使用MyEclipse來完成。

2、資料庫表建立

先安裝好MySQL資料庫。

安裝時，註意編碼選擇gbk。

通過控制台的mysql -u用戶名 –p密碼即可登錄mysql資料庫。

之後使用show databases可以看到所有的資料庫。

使用create database 可以建立一個新的庫。

使用 use 庫名，可以切換到另一個庫。

使用show tables可以看到一個庫下的所有表。

之後就可以通過普通的sql語句來建立表和進行數據的操作了。

在進行資料庫操作時，企業開發中必定要使用DAO（Data Access Object）設計模式

組成如下：

1） DataBaseConnection：建立資料庫連接

2） VO：與表對應的數據對象

3） DAO介面：規範操作方法

4） DAOImpl：針對DAO介面進行方法實現。

5） Factory：用來建立DAO介面對象。

首先根據需求，將資料庫表建立出來，這裡只需建立一個簡單的news新聞表，用於存儲網路上爬取得數據。

1 CREATE TABLE news (
2     id                int                primary key ,
3     title            varchar(200)    not null,
4     description        text            ,
5     url                varchar(200)
6 );

3、使用DAO設計模式進行資料庫操作

根據上述的DAO設計模式，我們需要編寫相關操作類，來完成資料庫的操作。

【Ⅰ】　　資料庫連接類DataBaseConnection，需要導入jar包:

 1 package org.liky.sina.dbc;
 2 
 3 import java.sql.Connection;
 4 import java.sql.DriverManager;
 5 import java.sql.SQLException;
 6 
 7 /**
 8  * 連接資料庫mysql的sina_news
 9  * @author k04
10  *
11  */
12 public class DataBaseConnection {
　　//此處可以試著兩種表達載入類的方法
13 //    private static final String DBORIVER="org.git.mm.mysql.Driver";
14     private static final String DBORIVER="com.mysql.jdbc.Driver";
15     private static final String DBURL="jdbc:mysql://localhost:3306/sina_news";
16     private static final String DBUSER="root";
17     private static final String DBPASSWORD="admin";
18     
19     private Connection conn;
20     /**
21      * 創建資料庫連接
22      * @return
23      */
24     public Connection getConnection(){
25         try {
26             if(conn==null||conn.isClosed()){
27                 //建立一個新的連接
28                     Class.forName(DBORIVER);
29                     conn=DriverManager.getConnection(DBURL, DBUSER, DBPASSWORD);
30                     //System.out.println("success  to connect!");
31             }
32         }catch (ClassNotFoundException e) {
33             // TODO Auto-generated catch block
34             e.printStackTrace();                
35         } catch (SQLException e) {
36             // TODO Auto-generated catch block
37             e.printStackTrace();
38         }
39         return conn;
40     }
41     /*
42      * 關閉連接
43      */
44     public void close(){
45         if(conn!=null){
46             try{
47                 conn.close();
48             }catch(SQLException e){
49                 e.printStackTrace();
50             }
51         }
52     }
53 
54 }

【Ⅱ】表單元素的封裝類News，根據建表時設定的四種元素id,title,description,url，將網路爬取得內容完整的導入資料庫中，此處可以用shift+Alt+S在Eclipse快捷創建封裝類：

 1 package org.liky.sina.vo;
 2 /**
 3  * news封裝類
 4  * @author k04
 5  *
 6  */
 7 public class News {
 8     private Integer id;
 9     private String title;
10     private String description;
11     private String url;
12     
13     
14     public News() {
15     }
16     public News(Integer id,String title,String description,String url) {
17         this.id=id;
18         this.title=title;
19         this.description=description;
20         this.url=url;
21     }
22     
23     
24     public Integer getId() {
25         return id;
26     }
27     public void setId(Integer id) {
28         this.id = id;
29     }
30     public String getTitle() {
31         return title;
32     }
33     public void setTitle(String title) {
34         this.title = title;
35     }
36     public String getDescription() {
37         return description;
38     }
39     public void setDescription(String description) {
40         this.description = description;
41     }
42     public String getUrl() {
43         return url;
44     }
45     public void setUrl(String url) {
46         this.url = url;
47     }
48 }

【Ⅲ】編寫DAO介面INewsDAO，存放資料庫操作類的方法名：

 1 package org.liky.sina.dao;
 2 /**
 3  * 介面，呈現三個方法
 4  */
 5 import java.util.List;
 6 import org.liky.sina.vo.News;
 7 
 8 public interface INewsDAO {
 9     /**
10      * 添加數據
11      * @param news    要添加的對象
12      * @throws Exception
13      */
14     public void doCreate(News news)throws Exception;
15     /**
16      * 根據主鍵id查詢數據
17      * 
18      */
19     public News findById(int id)throws Exception;
20     /**
21      * 根據一組id查詢所有結果
22      * @param ids    所有要查詢的id
23      * @return    查詢到的數據
24      * 因為索引是根據熱詞查到一堆的id
25      */
26     public List<News> findByIds(int[] ids)throws Exception;
27     
28 }

【Ⅳ】DAO介面的實現類NewsDAOImp類：

 1 package org.liky.sina.dao.impl;
 2 /**
 3  * 繼承INewsDAO介面
 4  * 實現三個方法，插入數據，查找指定id數據，查找一組id數據
 5  */
 6 import java.sql.PreparedStatement;
 7 import java.sql.ResultSet;
 8 import java.util.ArrayList;
 9 import java.util.List;
10 
11 import org.liky.sina.dao.INewsDAO;
12 import org.liky.sina.dbc.DataBaseConnection;
13 import org.liky.sina.vo.News;
14 
15 public class NewsDAOImpl implements INewsDAO {
16     //聲明一個資料庫連接類對象
17     private DataBaseConnection dbc;
18     
19     
20     //構造器，參數為資料庫連接類對象
21     public NewsDAOImpl(DataBaseConnection dbc) {
22         this.dbc=dbc;
23         
24     }
25 
26     @Override
27     public void doCreate(News news) throws Exception {
28         // TODO Auto-generated method stub
29         String sql="INSERT INTO news (id,title,description,url) VALUES (?,?,?,?)";
30         PreparedStatement pst=dbc.getConnection().prepareStatement(sql);
31         //設置參數
32         pst.setInt(1, news.getId());
33         pst.setString(2, news.getTitle());
34         pst.setString(3, news.getDescription());
35         pst.setString(4, news.getUrl());
36         
37         pst.executeUpdate();
38         System.out.println("create success.");
39     }
40 
41     @Override
42     public News findById(int id) throws Exception {
43         // TODO Auto-generated method stub
44         String sql="SELECT id,title,description,url FROM news WHERE id = ?";
45         PreparedStatement pst=dbc.getConnection().prepareStatement(sql);
46         pst.setInt(1, id);
47         ResultSet rs=pst.executeQuery();
48         News news=null;
49         //將符合id的數據遍歷寫入news並返回
50         if(rs.next()){
51             news=new News();
52             news.setId(rs.getInt(1));
53             news.setTitle(rs.getString(2));
54             news.setDescription(rs.getString(3));
55             news.setUrl(rs.getString(4));
56         }
57         //System.out.println("find success.");
58         return news;
59     }
60 
61     @Override
62     public List<News> findByIds(int[] ids) throws Exception {
63         // TODO Auto-generated method stub
64         StringBuilder sql=new StringBuilder("SELECT id,title,description,url FROM news WHERE id IN (");
65         //將id寫入ids，並用逗號隔開
66         if(ids!=null&&ids.length>0){
67             for(int id:ids){
68                 sql.append(id);
69                 sql.append(",");
70             }
71             //截取最後一個逗號，並補上括弧
72             String resultSQL=sql.substring(0, sql.length()-1)+")";
73             
74             PreparedStatement pst=dbc.getConnection().prepareStatement(resultSQL);
75             ResultSet rs=pst.executeQuery();
76             //存取一組id到鏈表中
77             List<News> list=new ArrayList<>();
78             while(rs.next()){
79                 News news=new News();
80                 news.setId(rs.getInt(1));
81                 news.setTitle(rs.getString(2));
82                 news.setDescription(rs.getString(3));
83                 news.setUrl(rs.getString(4));
84                 list.add(news);
85             }
86         }
87         //System.out.println("find success.");
88         return null;
89     }
90 
91 }

【Ⅴ】工廠類DAOFactory類，此類寫入了資料庫連接類參數，返回DAO實現類對象：

　　java中，我們通常有以下幾種創建對象的方式：

(1) 使用new關鍵字直接創建對象；

(2) 通過反射機制創建對象；

(3) 通過clone()方法創建對象；

(4) 通過工廠類創建對象。

 1 package org.liky.sina.factory;
 2 /**
 3  * 工廠類
 4  * 輸入一個連接資料庫對象的參數，返回資料庫表操作的類
 5  */
 6 import org.liky.sina.dao.INewsDAO;
 7 import org.liky.sina.dao.impl.NewsDAOImpl;
 8 import org.liky.sina.dbc.DataBaseConnection;
 9 
10 public class DAOFactory {
11     public static INewsDAO getINewsDAOInstance(DataBaseConnection dbc){
12         return new NewsDAOImpl(dbc);
13     }
14 }

4、網路爬蟲實現

　　現在編寫整個項目的重點，編寫URLDemo類，在爬蟲中進行資料庫的操作以及HDFS的寫入：

　　a' 關於此類，在網頁解析時用了簡單的Jsoup，並沒有如《java網路爬蟲》用正則表達式，所以需要導入jsoup的jar包；

　　b' 關於HDFS在eclipse的配置以及本機的連接，我後續博客會闡述，也可以網路查詢方法；

　　c' 這個類也是執行類，我收集的是新浪新聞網的數據，爬取深度為5，設置線程數5，並且篩選了只有鏈接含有“sian.news.com.cn”的。

　　d' 網路爬蟲我講了兩種方法：（1）java代碼實現網路爬蟲

　　　　　　　　　　　　　　 （2）Heritrix工具實現網路爬蟲

　　此處我還是選擇了直接寫代碼實現，自由度高也方便讀寫存取。

  1 package org.liky.sina.craw;
  2 
  3 import java.util.ArrayList;
  4 import java.util.HashMap;
  5 import java.util.HashSet;
  6 import java.util.List;
  7 import java.util.Map;
  8 import java.util.Set;
  9 
 10 import org.apache.hadoop.conf.Configuration;
 11 import org.apache.hadoop.fs.FSDataOutputStream;
 12 import org.apache.hadoop.fs.FileSystem;
 13 import org.apache.hadoop.fs.Path;
 14 import org.jsoup.Jsoup;
 15 import org.jsoup.nodes.Document;
 16 import org.jsoup.nodes.Element;
 17 import org.jsoup.select.Elements;
 18 import org.liky.sina.dao.INewsDAO;
 19 import org.liky.sina.dbc.DataBaseConnection;
 20 import org.liky.sina.factory.DAOFactory;
 21 import org.liky.sina.vo.News;
 22 
 23 /**
 24  * 爬蟲開始進行資料庫操作以及HDFS寫入
 25  * 
 26  * @author k04
 27  * 
 28  */
 29 public class URLDemo {
 30     // 該對象的構造方法會預設載入hadoop中的兩個配置文件，hdfs-site.xml和core-site.xml
 31     // 這兩個文件包含訪問hdfs所需的參數值
 32     private static Configuration conf = new Configuration();
 33 
 34     private static int id = 1;
 35 
 36     private static FileSystem fs;
 37 
 38     private static Path path;
 39 
 40     // 等待爬取的url
 41     private static List<String> allWaitUrl = new ArrayList<>();
 42     // 已經爬取的url
 43     private static Set<String> allOverUrl = new HashSet<>();
 44     // 記錄所有url的深度，以便在addUrl方法內判斷
 45     private static Map<String, Integer> allUrlDepth = new HashMap<>();
 46     // 爬取網頁的深度
 47     private static int maxDepth = 5;
 48     // 聲明object獨享幫助進行線程的等待操作
 49     private static Object obj = new Object();
 50     // 設置匯流排程數
 51     private static final int MAX_THREAD = 20;
 52     // 記錄空閑的線程數
 53     private static int count = 0;
 54 
 55     // 聲明INewsDAO對象，
 56     private static INewsDAO dao;
 57 
 58     static {
 59         dao = DAOFactory.getINewsDAOInstance(new DataBaseConnection());
 60     }
 61 
 62     public static void main(String args[]) {
 63         // 爬取的目標網址
 64         String strUrl = "http://news.sina.com.cn/";
 65 
 66         // 爬取第一個輸入的url
 67         addUrl(strUrl, 0);
 68         // 建立多個線程
 69         for (int i = 0; i < MAX_THREAD; i++) {
 70             new URLDemo().new MyThread().start();
 71         }
 72 
 73         // DataBaseConnection dc=new DataBaseConnection();
 74         // dc.getConnection();
 75 
 76     }
 77 
 78     public static void parseUrl(String strUrl, int depth) {
 79         // 先判斷當前url是否爬取過
 80         // 判斷深度是否符合要求
 81         if (!(allOverUrl.contains(strUrl) || depth > maxDepth)) {
 82             System.out.println("當前執行的  " + Thread.currentThread().getName()
 83                     + "  爬蟲線程處理爬取： " + strUrl);
 84 
 85             try {
 86                 // 用jsoup進行數據爬取
 87                 Document doc = Jsoup.connect(strUrl).get();
 88                 // 通過doc接受返回的結果
 89                 // 提取有效的title和description
 90                 String title = doc.title();
 91                 Element descE = doc.getElementsByAttributeValue("name",
 92                         "description").first();
 93                 String desc = descE.attr("content");
 94 
 95                 // System.out.println(title + " --> " + desc);
 96 
 97                 // 如果有效，則驚醒保存
 98                 if (title != null && desc != null && !title.trim().equals("")
 99                         && !desc.trim().equals("")) {
100                     // 需要生成一個id，以便放入資料庫中，因此id也要加入到HDFS中，便於後續索引
101                     News news = new News();
102                     news.setId(id++);
103                     news.setTitle(title);
104                     news.setDescription(desc);
105                     news.setUrl(strUrl);
106                     // 添加到資料庫語句
107                     dao.doCreate(news);
108                     // 向HDFS保存數據
109                     path = new Path("hdfs://localhost:9000/sina_news_input/"
110                             + System.currentTimeMillis() + ".txt");
111                     fs = path.getFileSystem(conf);
112                     FSDataOutputStream os = fs.create(path);
113                     // 進行內容輸出，此處需要用news.getId()，不然資料庫和HDFS的id會不相同，因為多線程的運行
114                     os.writeUTF(news.getId() + "\r\n" + title + "\r\n" + desc);
115                     os.close();
116 
117                     // 解析所有超鏈接
118                     Elements aEs = doc.getElementsByTag("a");
119                     // System.out.println(aEs);
120                     if (aEs != null && aEs.size() > 0) {
121                         for (Element aE : aEs) {
122                             String href = aE.attr("href");
123                             System.out.println(href);
124                             // 截取網址，並給出篩選條件！！！
125                             if ((href.startsWith("http:") || href
126                                     .startsWith("https:"))
127                                     && href.contains("news.sina.com.cn")) {
128                                 // 調用addUrl()方法
129                                 addUrl(href, depth + 1);
130                             }
131                         }
132                     }
133 
134                 }
135 
136             } catch (Exception e) {
137 
138             }
139             // 吧當前爬完的url放入到偶爾中
140             allOverUrl.add(strUrl);
141             System.out.println(strUrl + "爬去完成，已經爬取的內容量為：" + allOverUrl.size()
142                     + "剩餘爬取量為：" + allWaitUrl.size());
143 
144             // 判斷是否集合中海油其他的內容需要進行爬取，如果有，則進行線程的喚醒
145             if (allWaitUrl.size() > 0) {
146                 synchronized (obj) {
147                     obj.notify();
148                 }
149             } else {
150                 System.out.println("爬取結束...");
151                 System.exit(0);
152             }
153 
154         }
155     }
156 
157     /**
158      * url加入到等待隊列中 並判斷是否已經放過，若沒