作者：閑歡

來源：Python 技術(shù)

在上一篇文章《神器！五分鐘完成大型爬蟲項(xiàng)目！》，我們介紹了一個(gè)類似于 Scrapy 的開源爬蟲框架——feapder，并著重介紹了該框架的一種應(yīng)用——AirSpider，它是一個(gè)輕量級的爬蟲。

接下來我們再來介紹另一種爬蟲應(yīng)用——Spider，它是是一款基于 redis 的分布式爬蟲，適用于海量數(shù)據(jù)采集，支持?jǐn)帱c(diǎn)續(xù)爬、爬蟲報(bào)警、數(shù)據(jù)自動(dòng)入庫等功能。

安裝

和 AirSpider 一樣，我們也是通過命令行安裝。

由于 Spider 是分布式爬蟲，可能涉及到多個(gè)爬蟲，所以最好以項(xiàng)目的方式來創(chuàng)建。

創(chuàng)建項(xiàng)目

我們首先來創(chuàng)建項(xiàng)目：

feapder create -p spider-project

創(chuàng)建的項(xiàng)目目錄是這樣的：

創(chuàng)建好項(xiàng)目后，開發(fā)時(shí)我們需要將項(xiàng)目設(shè)置為工作區(qū)間，否則引入非同級目錄下的文件時(shí)，編譯器會(huì)報(bào)錯(cuò)。

設(shè)置工作區(qū)間方式（以pycharm為例）：項(xiàng)目->右鍵->Mark Directory as -> Sources Root。

創(chuàng)建爬蟲

創(chuàng)建爬蟲的命令行語句為：

feapder create -s <spider_name> <spider_type>

• AirSpider 對應(yīng)的 spider_type 值為 1
• Spider 對應(yīng)的 spider_type 值為 2
• BatchSpider 對應(yīng)的 spider_type 值為 3

默認(rèn) spider_type 值為 1。

所以創(chuàng)建 Spider 的語句為：

feapder create -s spider_test 2

運(yùn)行語句后，我們可以看到在 spiders 目錄下生成了 spider_test.py 文件。

對應(yīng)的文件內(nèi)容為：

import feapder class SpiderTest(feapder.Spider): # 自定義數(shù)據(jù)庫，若項(xiàng)目中有setting.py文件，此自定義可刪除 __custom_setting__ = dict(
        REDISDB_IP_PORTS="localhost:6379", REDISDB_USER_PASS="", REDISDB_DB=0 ) def start_requests(self): yield feapder.Request("https://www.baidu.com") def parse(self, request, response): print(response) if __name__ == "__main__":
    SpiderTest(redis_key="xxx:xxx").start()

因Spider是基于redis做的分布式，因此模板代碼默認(rèn)給了redis的配置方式。關(guān)于 Redis 的配置信息：

• REDISDB_IP_PORTS：連接地址，若為集群或哨兵模式，多個(gè)連接地址用逗號(hào)分開，若為哨兵模式，需要加個(gè)REDISDB_SERVICE_NAME參數(shù)。
• REDISDB_USER_PASS：連接密碼。
• REDISDB_DB：數(shù)據(jù)庫。

在 main 函數(shù)中，我們可以看到有個(gè) redis_key 的參數(shù)，這個(gè)參數(shù)是redis中存儲(chǔ)任務(wù)等信息的 key 前綴，如 redis_key="feapder:spider_test", 則redis中會(huì)生成如下：

特性

我們在 AirSpider 里面講的方法，在 Spider 這里都支持，下面我們來看看 Spider 相對于 AirSpider 的不同之處。

數(shù)據(jù)自動(dòng)入庫

寫過爬蟲的人都知道，如果要將數(shù)據(jù)持久化到 MySQL 數(shù)據(jù)庫，如果碰到字段特別多的情況，就會(huì)很煩人，需要解析之后手寫好多字段，拼湊 SQL 語句。

這個(gè)問題，Spider 幫我們想到了，我們可以利用框架幫我們自動(dòng)入庫。

建表

第一步，我們需要在數(shù)據(jù)庫中創(chuàng)建一張數(shù)據(jù)表，這個(gè)大家都會(huì)，這里就不說了。

配置 setting

將 setting.py 里面的數(shù)據(jù)庫配置改為自己的配置：

# # MYSQL MYSQL_IP = "" MYSQL_PORT = MYSQL_DB = "" MYSQL_USER_NAME = "" MYSQL_USER_PASS = "" 

也就是這幾個(gè)配置。

生成實(shí)體類 Item

接著，我們將命令行切換到我們項(xiàng)目的 items 目錄，運(yùn)行命令：

feapder create -i <item_name>

我這里數(shù)據(jù)庫里使用的是 report 表，所以命令為：

feapder create -i report

然后，我們就可以在 items 目錄下看到生成的 report_item.py 實(shí)體類了。我這里生成的實(shí)體類內(nèi)容是：

from feapder import Item class ReportItem(Item): """
    This class was generated by feapder.
    command: feapder create -i report.
    """ __table_name__ = "report" def __init__(self, *args, **kwargs): self.count = None self.emRatingName = None # 評級名稱 self.emRatingValue = None # 評級代碼 self.encodeUrl = None # 鏈接 # self.id = None self.indvInduCode = None # 行業(yè)代碼 self.indvInduName = None # 行業(yè)名稱 self.lastEmRatingName = None # 上次評級名稱 self.lastEmRatingValue = None # 上次評級代碼 self.orgCode = None # 機(jī)構(gòu)代碼 self.orgName = None # 機(jī)構(gòu)名稱 self.orgSName = None # 機(jī)構(gòu)簡稱 self.predictNextTwoYearEps = None self.predictNextTwoYearPe = None self.predictNextYearEps = None self.predictNextYearPe = None self.predictThisYearEps = None self.predictThisYearPe = None self.publishDate = None # 發(fā)表時(shí)間 self.ratingChange = None # 評級變動(dòng) self.researcher = None # 研究員 self.stockCode = None # 股票代碼 self.stockName = None # 股票簡稱 self.title = None # 報(bào)告名稱 

若字段有默認(rèn)值或者自增，則默認(rèn)注釋掉，可按需打開。大家可以看到我這張表的 id 字段在這里被注釋了。

若item字段過多，不想逐一賦值，可通過如下方式創(chuàng)建：

feapder create -i report 1

這時(shí)候生成的實(shí)體類是這樣的：

class ReportItem(Item): """
    This class was generated by feapder.
    command: feapder create -i report 1.
    """ __table_name__ = "report 1" def __init__(self, *args, **kwargs): self.count = kwargs.get('count') self.emRatingName = kwargs.get('emRatingName') # 評級名稱 self.emRatingValue = kwargs.get('emRatingValue') # 評級代碼 self.encodeUrl = kwargs.get('encodeUrl') # 鏈接 # self.id = kwargs.get('id') self.indvInduCode = kwargs.get('indvInduCode') # 行業(yè)代碼 self.indvInduName = kwargs.get('indvInduName') # 行業(yè)名稱 self.lastEmRatingName = kwargs.get('lastEmRatingName') # 上次評級名稱 self.lastEmRatingValue = kwargs.get('lastEmRatingValue') # 上次評級代碼 self.orgCode = kwargs.get('orgCode') # 機(jī)構(gòu)代碼 self.orgName = kwargs.get('orgName') # 機(jī)構(gòu)名稱 self.orgSName = kwargs.get('orgSName') # 機(jī)構(gòu)簡稱 self.predictNextTwoYearEps = kwargs.get('predictNextTwoYearEps') self.predictNextTwoYearPe = kwargs.get('predictNextTwoYearPe') self.predictNextYearEps = kwargs.get('predictNextYearEps') self.predictNextYearPe = kwargs.get('predictNextYearPe') self.predictThisYearEps = kwargs.get('predictThisYearEps') self.predictThisYearPe = kwargs.get('predictThisYearPe') self.publishDate = kwargs.get('publishDate') # 發(fā)表時(shí)間 self.ratingChange = kwargs.get('ratingChange') # 評級變動(dòng) self.researcher = kwargs.get('researcher') # 研究員 self.stockCode = kwargs.get('stockCode') # 股票代碼 self.stockName = kwargs.get('stockName') # 股票簡稱 self.title = kwargs.get('title') # 報(bào)告名稱 

這樣當(dāng)我們請求回來的json數(shù)據(jù)時(shí)，可直接賦值，如:

response_data = {"title":" 測試"} # 模擬請求回來的數(shù)據(jù) item = SpiderDataItem(**response_data)

想要數(shù)據(jù)自動(dòng)入庫也比較簡單，在解析完數(shù)據(jù)之后，將數(shù)據(jù)賦值給 Item，然后 yield 就行了：

def parse(self, request, response):
        html = response.content.decode("utf-8")
        if len(html):
            content = html.replace('datatable1351846(', '')[:-1]
            content_json = json.loads(content)
            print(content_json) for obj in content_json['data']: result = ReportItem() result['orgName'] = obj['orgName'] #機(jī)構(gòu)名稱 result['orgSName'] = obj['orgSName'] #機(jī)構(gòu)簡稱 result['publishDate'] = obj['publishDate'] #發(fā)布日期 result['predictNextTwoYearEps'] = obj['predictNextTwoYearEps'] #后年每股盈利 result['title'] = obj['title'] #報(bào)告名稱 result['stockName'] = obj['stockName'] #股票名稱 result['stockCode'] = obj['stockCode'] #股票code result['orgCode'] = obj['stockCode'] #機(jī)構(gòu)code result['predictNextTwoYearPe'] = obj['predictNextTwoYearPe'] #后年市盈率 result['predictNextYearEps'] = obj['predictNextYearEps'] # 明年每股盈利 result['predictNextYearPe'] = obj['predictNextYearPe'] # 明年市盈率 result['predictThisYearEps'] = obj['predictThisYearEps'] #今年每股盈利 result['predictThisYearPe'] = obj['predictThisYearPe'] #今年市盈率 result['indvInduCode'] = obj['indvInduCode'] # 行業(yè)代碼 result['indvInduName'] = obj['indvInduName'] # 行業(yè)名稱 result['lastEmRatingName'] = obj['lastEmRatingName'] # 上次評級名稱 result['lastEmRatingValue'] = obj['lastEmRatingValue'] # 上次評級代碼 result['emRatingValue'] = obj['emRatingValue'] # 評級代碼 result['emRatingName'] = obj['emRatingName'] # 評級名稱 result['ratingChange'] = obj['ratingChange'] # 評級變動(dòng) result['researcher'] = obj['researcher'] # 研究員 result['encodeUrl'] = obj['encodeUrl'] # 鏈接 result['count'] = int(obj['count']) # 近一月個(gè)股研報(bào)數(shù) yield result 

返回item后，item 會(huì)流進(jìn)到框架的 ItemBuffer, ItemBuffer 每.05秒或當(dāng)item數(shù)量積攢到5000個(gè)，便會(huì)批量將這些 item 批量入庫。表名為類名去掉 Item 的小寫，如 ReportItem 數(shù)據(jù)會(huì)落入到 report 表。

調(diào)試

開發(fā)過程中，我們可能需要針對某個(gè)請求進(jìn)行調(diào)試，常規(guī)的做法是修改下發(fā)任務(wù)的代碼。但這樣并不好，改來改去可能把之前寫好的邏輯搞亂了，或者忘記改回來直接發(fā)布了，又或者調(diào)試的數(shù)據(jù)入庫了，污染了庫里已有的數(shù)據(jù)，造成了很多本來不應(yīng)該發(fā)生的問題。

本框架支持Debug爬蟲，可針對某條任務(wù)進(jìn)行調(diào)試，寫法如下：

if __name__ == "__main__":
    spider = SpiderTest.to_DebugSpider(
        redis_key="feapder:spider_test", request=feapder.Request("http://www.baidu.com")
    )
    spider.start()

對比之前的啟動(dòng)方式：

spider = SpiderTest(redis_key="feapder:spider_test")
spider.start()

可以看到，代碼中 to_DebugSpider 方法可以將原爬蟲直接轉(zhuǎn)為 debug 爬蟲，然后通過傳遞 request 參數(shù)抓取指定的任務(wù)。

通常結(jié)合斷點(diǎn)來進(jìn)行調(diào)試，debug 模式下，運(yùn)行產(chǎn)生的數(shù)據(jù)默認(rèn)不入庫。

除了指定 request 參數(shù)外，還可以指定 request_dict 參數(shù)，request_dict 接收字典類型，如 request_dict={"url":"http://www.baidu.com"}, 其作用于傳遞 request 一致。request 與 request_dict 二者選一傳遞即可。

運(yùn)行多個(gè)爬蟲

通常，一個(gè)項(xiàng)目下可能存在多個(gè)爬蟲，為了規(guī)范，建議啟動(dòng)入口統(tǒng)一放到項(xiàng)目下的 main.py 中，然后以命令行的方式運(yùn)行指定的文件。

例如如下項(xiàng)目：

項(xiàng)目中包含了兩個(gè)spider，main.py寫法如下：

from spiders import * from feapder import Request from feapder import ArgumentParser def test_spider(): spider = test_spider.TestSpider(redis_key="spider:report")
    spider.start() def test_spider2(): spider = test_spider.TestSpider2(redis_key="spider:report")
    spider.start() if __name__ == "__main__":
    parser = ArgumentParser(description="Spider測試")

    parser.add_argument( "--test_spider", action="store_true", help="測試Spider", function=test_spider
    )
    parser.add_argument( "--test_spider2", action="store_true", help="測試Spider2", function=test_spider2
    )
   
    parser.start()

這里使用了 ArgumentParser 模塊，使其支持命令行參數(shù)，如運(yùn)行 test_spider:

python3 main.py --test_spider

分布式

分布式說白了就是啟動(dòng)多個(gè)進(jìn)程，處理同一批任務(wù)。Spider 支持啟動(dòng)多份，且不會(huì)重復(fù)發(fā)下任務(wù)，我們可以在多個(gè)服務(wù)器上部署啟動(dòng)，也可以在同一個(gè)機(jī)器上啟動(dòng)多次。

總結(jié)

到這里， Spider 分布式爬蟲咱就講完了，還有一些細(xì)節(jié)的東西，大家在用的時(shí)候還需要琢磨一下?？傮w來說，這個(gè)框架還是比較好用的，上手簡單，應(yīng)對一些不是很復(fù)雜的場景綽綽有余，大家可以嘗試著將自己的爬蟲重構(gòu)一下，試試這款框架。