python簡單分佈式爬蟲

本章講的依舊是實戰項目，實戰內容是打造分佈式爬蟲，這對初學者來講，是一個不小的挑戰，也是一次有意義的嘗試。此次打造的分佈式爬蟲採用比較簡單的主從模式，徹底手工打造，不使用成熟框架，基本上涵蓋了前六章的主要知識點，其中涉及分佈式的知識點是分佈式進程和進程間通訊的內容，算是對Python爬蟲基礎篇的總結。

　　如今大型的爬蟲系統都是採起分佈式爬取結構，經過這次實戰項目，讓你們對分佈式爬蟲有一個比較清晰地瞭解，爲以後系統的講解分佈式爬蟲打下基礎，其實它並無多麼困難。實戰目標：爬取2000個百度百科網絡爬蟲詞條以及相關詞條的標題、摘要和連接等信息，採用分佈式結構改寫第六章的基礎爬蟲，使功能更增強大。爬取頁面請看圖6.1。

                       

 

7.1簡單分佈式爬蟲結構

　　本次分佈式爬蟲採用主從模式。主從模式是指由一臺主機做爲控制節點負責全部運行網絡爬蟲的主機進行管理，爬蟲只須要從控制節點那裏接收任務，並把新生成任務提交給控制節點就能夠了，在這個過程當中沒必要與其餘爬蟲通訊，這種方式實現簡單利於管理。而控制節點則須要與全部爬蟲進行通訊，所以能夠看到主從模式是有缺陷的，控制節點會成爲整個系統的瓶頸，容易致使整個分佈式網絡爬蟲系統性能降低。

這次使用三臺主機進行分佈式爬取，一臺主機做爲控制節點，另外兩臺主機做爲爬蟲節點。爬蟲結構如圖7.1所示：

                        

圖7.1 主從爬蟲結構

 

7.2 控制節點ControlNode

　　控制節點主要分爲URL管理器、數據存儲器和控制調度器。控制調度器經過三個進程來協調URL管理器和數據存儲器的工做，一個是URL管理進程，負責URL的管理和將URL傳遞給爬蟲節點，一個是數據提取進程，負責讀取爬蟲節點返回的數據，將返回數據中的URL交給URL管理進程，將標題和摘要等數據交給數據存儲進程，最後一個是數據存儲進程，負責將數據提取進程中提交的數據進行本地存儲。執行流程如圖7.2所示：

                      

圖7.2 控制節點執行流程

7.2.1 URL管理器

　　URL管理器查考第六章的代碼，作了一些優化修改。因爲咱們採用set內存去重的方式，若是直接存儲大量的URL連接，尤爲是URL連接很長的時候，很容易形成內存溢出，因此咱們採用將爬取過的URL進行MD5處理，因爲字符串通過MD5處理後的信息摘要長度能夠128bit，將生成的MD5摘要存儲到set後，能夠減小好幾倍的內存消耗，Python中的MD5算法生成的是32位的字符串，因爲咱們爬取的url較少，md5衝突不大，徹底能夠取中間的16位字符串，即16位MD5加密。同時添加了save_progress和load_progress方法進行序列化的操做，將未爬取URL集合和已爬取的URL集合序列化到本地，保存當前的進度，以便下次恢復狀態。URL管理器URLManager.py代碼以下：

 1 #coding:utf-8
 2 import cPickle
 3 import hashlib
 4 class UrlManager(object):
 5     def __init__(self):
 6         self.new_urls = self.load_progress('new_urls.txt')#未爬取URL集合
 7         self.old_urls = self.load_progress('old_urls.txt')#已爬取URL集合
 8     def has_new_url(self):
 9         '''
10         判斷是否有未爬取的URL
11         :return:
12         '''
13         return self.new_url_size()!=0
14  
15     def get_new_url(self):
16         '''
17         獲取一個未爬取的URL
18         :return:
19         '''
20         new_url = self.new_urls.pop()
21         m = hashlib.md5()
22         m.update(new_url)
23         self.old_urls.add(m.hexdigest()[8:-8])
24         return new_url
25  
26     def add_new_url(self,url):
27         '''
28          將新的URL添加到未爬取的URL集合中
29         :param url:單個URL
30         :return:
31         '''
32         if url is None:
33             return
34         m = hashlib.md5()
35         m.update(url)
36         url_md5 =  m.hexdigest()[8:-8]
37         if url not in self.new_urls and url_md5 not in self.old_urls:
38             self.new_urls.add(url)
39  
40     def add_new_urls(self,urls):
41         '''
42         將新的URLS添加到未爬取的URL集合中
43         :param urls:url集合
44         :return:
45         '''
46         if urls is None or len(urls)==0:
47             return
48         for url in urls:
49             self.add_new_url(url)
50  
51     def new_url_size(self):
52         '''
53         獲取未爬取URL集合的s大小
54         :return:
55         '''
56         return len(self.new_urls)
57  
58     def old_url_size(self):
59         '''
60         獲取已經爬取URL集合的大小
61         :return:
62         '''
63         return len(self.old_urls)
64  
65     def save_progress(self,path,data):
66         '''
67         保存進度
68         :param path:文件路徑
69         :param data:數據
70         :return:
71         '''
72         with open(path, 'wb') as f:
73             cPickle.dump(data, f)
74  
75     def load_progress(self,path):
76         '''
77         從本地文件加載進度
78         :param path:文件路徑
79         :return:返回set集合
80         '''
81         print '[+] 從文件加載進度: %s' % path
82         try:
83             with open(path, 'rb') as f:
84                 tmp = cPickle.load(f)
85                 return tmp
86         except:
87             print '[!] 無進度文件, 建立: %s' % path
88         return set()

 

7.2.2數據存儲器

數據存儲器的內容基本上和第六章的同樣，不過生成文件按照當前時間進行命名避免重複，同時對文件進行緩存寫入。代碼以下：

 1 #coding:utf-8
 2 import codecs
 3 import time
 4 class DataOutput(object):
 5     def __init__(self):
 6         self.filepath='baike_%s.html'%(time.strftime("%Y_%m_%d_%H_%M_%S", time.localtime()) )
 7         self.output_head(self.filepath)
 8         self.datas=[]
 9     def store_data(self,data):
10         if data is None:
11             return
12         self.datas.append(data)
13         if len(self.datas)>10:
14             self.output_html(self.filepath)
15  
16  
17     def output_head(self,path):
18         '''
19         將HTML頭寫進去
20         :return:
21         '''
22         fout=codecs.open(path,'w',encoding='utf-8')
23         fout.write("<html>")
24         fout.write("<body>")
25         fout.write("<table>")
26         fout.close()
27  
28     def output_html(self,path):
29         '''
30         將數據寫入HTML文件中
31         :param path: 文件路徑
32         :return:
33         '''
34         fout=codecs.open(path,'a',encoding='utf-8')
35         for data in self.datas:
36             fout.write("<tr>")
37             fout.write("<td>%s</td>"%data['url'])
38             fout.write("<td>%s</td>"%data['title'])
39             fout.write("<td>%s</td>"%data['summary'])
40             fout.write("</tr>")
41             self.datas.remove(data)
42         fout.close()
43  
44     def ouput_end(self,path):
45         '''
46         輸出HTML結束
47         :param path: 文件存儲路徑
48         :return:
49         '''
50         fout=codecs.open(path,'a',encoding='utf-8')
51         fout.write("</table>")
52         fout.write("</body>")
53         fout.write("</html>")
54         fout.close()

 

7.2.3控制調度器

　　控制調度器主要是產生並啓動URL管理進程、數據提取進程和數據存儲進程，同時維護4個隊列保持進程間的通訊，分別爲url_queue,result_queue, conn_q,store_q。4個隊列說明以下：

• url_q隊列是URL管理進程將URL傳遞給爬蟲節點的通道
• result_q隊列是爬蟲節點將數據返回給數據提取進程的通道
• conn_q隊列是數據提取進程將新的URL數據提交給URL管理進程的通道
• store_q隊列是數據提取進程將獲取到的數據交給數據存儲進程的通道

由於要和工做節點進行通訊，因此分佈式進程必不可少。參考1.4.4小節分佈式進程中服務進程中的代碼(linux版)，建立一個分佈式管理器，定義爲start_manager方法。方法代碼以下：

 1 def start_Manager(self,url_q,result_q):
 2     '''
 3     建立一個分佈式管理器
 4     :param url_q: url隊列
 5     :param result_q: 結果隊列
 6     :return:
 7     '''
 8     #把建立的兩個隊列註冊在網絡上，利用register方法，callable參數關聯了Queue對象，
 9     # 將Queue對象在網絡中暴露
10     BaseManager.register('get_task_queue',callable=lambda:url_q)
11     BaseManager.register('get_result_queue',callable=lambda:result_q)
12     #綁定端口8001，設置驗證口令‘baike’。這個至關於對象的初始化
13     manager=BaseManager(address=('',8001),authkey='baike')
14     #返回manager對象
15     return manager

 

 

　　

URL管理進程將從conn_q隊列獲取到的新URL提交給URL管理器，通過去重以後，取出URL放入url_queue隊列中傳遞給爬蟲節點，代碼以下：

def url_manager_proc(self,url_q,conn_q,root_url):
    url_manager = UrlManager()
    url_manager.add_new_url(root_url)
    while True:
        while(url_manager.has_new_url()):
 
            #從URL管理器獲取新的url
            new_url = url_manager.get_new_url()
            #將新的URL發給工做節點
            url_q.put(new_url)
            print 'old_url=',url_manager.old_url_size()
            #加一個判斷條件，當爬去2000個連接後就關閉,並保存進度
            if(url_manager.old_url_size()>2000):
                #通知爬行節點工做結束
                url_q.put('end')
                print '控制節點發起結束通知!'
                #關閉管理節點，同時存儲set狀態
                url_manager.save_progress('new_urls.txt',url_manager.new_urls)
                url_manager.save_progress('old_urls.txt',url_manager.old_urls)
                return
        #將從result_solve_proc獲取到的urls添加到URL管理器之間
        try:
            if not conn_q.empty():
                urls = conn_q.get()
                url_manager.add_new_urls(urls)
        except BaseException,e:
            time.sleep(0.1)#延時休息

 

　　數據提取進程從result_queue隊列讀取返回的數據，並將數據中的URL添加到conn_q隊列交給URL管理進程，將數據中的文章標題和摘要添加到store_q隊列交給數據存儲進程。代碼以下：

def result_solve_proc(self,result_q,conn_q,store_q):
    while(True):
        try:
            if not result_q.empty():
                content = result_q.get(True)
                if content['new_urls']=='end':
                    #結果分析進程接受通知而後結束
                    print '結果分析進程接受通知而後結束!'
                    store_q.put('end')
                    return
                conn_q.put(content['new_urls'])#url爲set類型
                store_q.put(content['data'])#解析出來的數據爲dict類型
            else:
                time.sleep(0.1)#延時休息
        except BaseException,e:
            time.sleep(0.1)#延時休息

 

數據存儲進程從store_q隊列中讀取數據，並調用數據存儲器進行數據存儲。代碼以下：

 1 def store_proc(self,store_q):
 2     output = DataOutput()
 3     while True:
 4         if not store_q.empty():
 5             data = store_q.get()
 6             if data=='end':
 7                 print '存儲進程接受通知而後結束!'
 8                 output.ouput_end(output.filepath)
 9  
10                 return
11             output.store_data(data)
12         else:
13             time.sleep(0.1)

 

最後將分佈式管理器、URL管理進程、 數據提取進程和數據存儲進程進行啓動，並初始化4個隊列。代碼以下：

 1 if __name__=='__main__':
 2     #初始化4個隊列
 3     url_q = Queue()
 4     result_q = Queue()
 5     store_q = Queue()
 6     conn_q = Queue()
 7     #建立分佈式管理器
 8     node = NodeManager()
 9     manager = node.start_Manager(url_q,result_q)
10     #建立URL管理進程、 數據提取進程和數據存儲進程
11     url_manager_proc = Process(target=node.url_manager_proc, args=(url_q,conn_q,'http://baike.baidu.com/view/284853.htm',))
12     result_solve_proc = Process(target=node.result_solve_proc, args=(result_q,conn_q,store_q,))
13     store_proc = Process(target=node.store_proc, args=(store_q,))
14     #啓動3個進程和分佈式管理器
15     url_manager_proc.start()
16     result_solve_proc.start()
17     store_proc.start()
18     manager.get_server().serve_forever()

 

 

　　

7.3 爬蟲節點SpiderNode

爬蟲節點相對簡單，主要包含HTML下載器、HTML解析器和爬蟲調度器。執行流程以下：

• 爬蟲調度器從控制節點中的url_q隊列讀取URL
• 爬蟲調度器調用HTML下載器、HTML解析器獲取網頁中新的URL和標題摘要
• 最後爬蟲調度器將新的URL和標題摘要傳入result_q隊列交給控制節點

7.3.1 HTML下載器

HTML下載器的代碼和第六章的一致，只要注意網頁編碼便可。代碼以下：

 1 #coding:utf-8
 2 import requests
 3 class HtmlDownloader(object):
 4  
 5     def download(self,url):
 6         if url is None:
 7             return None
 8         user_agent = 'Mozilla/4.0 (compatible; MSIE 5.5; Windows NT)'
 9         headers={'User-Agent':user_agent}
10         r = requests.get(url,headers=headers)
11         if r.status_code==200:
12             r.encoding='utf-8'
13             return r.text
14         return None

 

7.3.2 HTML解析器

HTML解析器的代碼和第六章的一致，詳細的網頁分析過程能夠回顧第六章。代碼以下：

 1 #coding:utf-8
 2 import re
 3 import urlparse
 4 from bs4 import BeautifulSoup
 5  
 6  
 7 class HtmlParser(object):
 8  
 9     def parser(self,page_url,html_cont):
10         '''
11         用於解析網頁內容抽取URL和數據
12         :param page_url: 下載頁面的URL
13         :param html_cont: 下載的網頁內容
14         :return:返回URL和數據
15         '''
16         if page_url is None or html_cont is None:
17             return
18         soup = BeautifulSoup(html_cont,'html.parser',from_encoding='utf-8')
19         new_urls = self._get_new_urls(page_url,soup)
20         new_data = self._get_new_data(page_url,soup)
21         return new_urls,new_data
22  
23  
24     def _get_new_urls(self,page_url,soup):
25         '''
26         抽取新的URL集合
27         :param page_url: 下載頁面的URL
28         :param soup:soup
29         :return: 返回新的URL集合
30         '''
31         new_urls = set()
32         #抽取符合要求的a標籤
33         links = soup.find_all('a',href=re.compile(r'/view/\d+\.htm'))
34         for link in links:
35             #提取href屬性
36             new_url = link['href']
37             #拼接成完整網址
38             new_full_url = urlparse.urljoin(page_url,new_url)
39             new_urls.add(new_full_url)
40         return new_urls
41     def _get_new_data(self,page_url,soup):
42         '''
43         抽取有效數據
44         :param page_url:下載頁面的URL
45         :param soup:
46         :return:返回有效數據
47         '''
48         data={}
49         data['url']=page_url
50         title = soup.find('dd',class_='lemmaWgt-lemmaTitle-title').find('h1')
51         data['title']=title.get_text()
52         summary = soup.find('div',class_='lemma-summary')
53         #獲取tag中包含的全部文版內容包括子孫tag中的內容,並將結果做爲Unicode字符串返回
54         data['summary']=summary.get_text()
55         return data

 

7.3.3 爬蟲調度器

　　爬蟲調度器須要用到分佈式進程中工做進程的代碼，具體內容能夠參考第一章的分佈式進程章節。爬蟲調度器須要先鏈接上控制節點，而後依次完成從url_q隊列中獲取URL，下載並解析網頁，將獲取的數據交給result_q隊列，返回給控制節點等各項任務，代碼以下：

 1 class SpiderWork(object):
 2     def __init__(self):
 3         #初始化分佈式進程中的工做節點的鏈接工做
 4         # 實現第一步：使用BaseManager註冊獲取Queue的方法名稱
 5         BaseManager.register('get_task_queue')
 6         BaseManager.register('get_result_queue')
 7         # 實現第二步：鏈接到服務器:
 8         server_addr = '127.0.0.1'
 9         print('Connect to server %s...' % server_addr)
10         # 端口和驗證口令注意保持與服務進程設置的徹底一致:
11         self.m = BaseManager(address=(server_addr, 8001), authkey='baike')
12         # 從網絡鏈接:
13         self.m.connect()
14         # 實現第三步：獲取Queue的對象:
15         self.task = self.m.get_task_queue()
16         self.result = self.m.get_result_queue()
17         #初始化網頁下載器和解析器
18         self.downloader = HtmlDownloader()
19         self.parser = HtmlParser()
20         print 'init finish'
21  
22     def crawl(self):
23         while(True):
24             try:
25                 if not self.task.empty():
26                     url = self.task.get()
27  
28                     if url =='end':
29                         print '控制節點通知爬蟲節點中止工做...'
30                         #接着通知其它節點中止工做
31                         self.result.put({'new_urls':'end','data':'end'})
32                         return
33                     print '爬蟲節點正在解析:%s'%url.encode('utf-8')
34                     content = self.downloader.download(url)
35                     new_urls,data = self.parser.parser(url,content)
36                     self.result.put({"new_urls":new_urls,"data":data})
37             except EOFError,e:
38                 print "鏈接工做節點失敗"
39                 return
40             except Exception,e:
41                 print e
42                 print 'Crawl  fali '
43  
44 if __name__=="__main__":
45     spider = SpiderWork()
46     spider.crawl()

 

　　在爬蟲調度器設置了一個本地IP：127.0.0.1，你們能夠將在一臺機器上測試代碼的正確性。固然也可使用三臺VPS服務器，兩臺運行爬蟲節點程序，將IP改成控制節點主機的公網IP，一臺運行控制節點程序，進行分佈式爬取，這樣更貼近真實的爬取環境。下面圖7.3爲最終爬取的數據，圖7.4爲new_urls.txt內容，圖7.5爲old_urls.txt內容，你們能夠進行對比測試，這個簡單的分佈式爬蟲還有很大發揮的空間，但願你們發揮本身的聰明才智進一步完善。

                                 

圖7.3 最終爬取的數據

                                  

圖7.4 new_urls.txt

                                    

圖7.5 old_urls.txt

7.4小結

　　本章講解了一個簡單的分佈式爬蟲結構，主要目的是幫助你們將Python爬蟲基礎篇的知識進行總結和強化，開拓你們的思惟，同時也讓你們知道分佈式爬蟲並非這麼遙不可及。不過當你親手打造一個分佈式爬蟲後，就會知道分佈式爬蟲的難點在於節點的調度，什麼樣的結構能讓各個節點穩定高效的運做纔是分佈式爬蟲要考慮的核心內容。到本章爲止，Python爬蟲基礎篇已經結束，這個時候你們基本上能夠編寫簡單的爬蟲，爬取一些靜態網站的內容，可是Python爬蟲開發不只如此，你們接着往下學習吧。

 

本文轉載自  http://www.cnblogs.com/qiyeboy/p/7016540.html，在此僅作學習用，謝謝！