第三百六十七節，Python分布式爬蟲打造搜索引擎Scrapy精講—elasticsearch(搜索引擎)scrapy寫入數據到elasticsearch中

第三百六十七節，Python分布式爬蟲打造搜索引擎Scrapy精講—elasticsearch(搜索引擎)scrapy寫入數據到elasticsearch中

 

前面我們講到的elasticsearch(搜索引擎)操作，如：增、刪、改、查等操作都是用的elasticsearch的語言命令，就像sql命令一樣，當然elasticsearch官方也提供了一個python操作elasticsearch(搜索引擎)的接口包，就像sqlalchemy操作數據庫一樣的ORM框，這樣我們操作elasticsearch就不用寫命令了，用elasticsearch-dsl-py這個模塊來操作，也就是用python的方式操作一個類即可

 

elasticsearch-dsl-py下載

下載地址：https://github.com/elastic/elasticsearch-dsl-py

文檔說明：http://elasticsearch-dsl.readthedocs.io/en/latest/

首先安裝好elasticsearch-dsl-py模塊

 

1、elasticsearch-dsl模塊使用說明

create_connection(hosts=['127.0.0.1'])：連接elasticsearch(搜索引擎)服務器方法，可以連接多台服務器
class Meta：設置索引名稱和表名稱
索引類名稱.init(): 生成索引和表以及字段
實例化索引類.save():將數據寫入elasticsearch(搜索引擎)

 

elasticsearch_orm.py 操作elasticsearch(搜索引擎)文件

#!/usr/bin/env python
# -*- coding:utf8 -*-
from datetime import datetime
from elasticsearch_dsl import DocType, Date, Nested, Boolean, \ analyzer, InnerObjectWrapper, Completion, Keyword, Text, Integer # 更多字段類型見第三百六十四節elasticsearch(搜索引擎)的mapping映射管理

from elasticsearch_dsl.connections import connections       # 導入連接elasticsearch(搜索引擎)服務器方法
connections.create_connection(hosts=['127.0.0.1']) class lagouType(DocType):                                                   # 自定義一個類來繼承DocType類
    # Text類型需要分詞，所以需要知道中文分詞器，ik_max_wordwei為中文分詞器
    title = Text(analyzer="ik_max_word")                                    # 設置，字段名稱=字段類型，Text為字符串類型並且可以分詞建立倒排索引
    description = Text(analyzer="ik_max_word") keywords = Text(analyzer="ik_max_word") url = Keyword()                                                         # 設置，字段名稱=字段類型，Keyword為普通字符串類型，不分詞
    riqi = Date()                                                           # 設置，字段名稱=字段類型，Date日期類型

    class Meta:                                                             # Meta是固定寫法
        index = "lagou"                                                     # 設置索引名稱(相當於數據庫名稱)
        doc_type = 'biao'                                                   # 設置表名稱

if __name__ == "__main__":          # 判斷在本代碼文件執行才執行里面的方法，其他頁面調用的則不執行里面的方法
    lagouType.init()                # 生成elasticsearch(搜索引擎)的索引，表，字段等信息


# 使用方法說明： # 在要要操作elasticsearch(搜索引擎)的頁面，導入此模塊 # lagou = lagouType() #實例化類 # lagou.title = '值' #要寫入字段=值 # lagou.description = '值' # lagou.keywords = '值' # lagou.url = '值' # lagou.riqi = '值' # lagou.save() #將數據寫入elasticsearch(搜索引擎)

 

 

2、scrapy寫入數據到elasticsearch中

爬蟲文件

# -*- coding: utf-8 -*-
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from adc.items import LagouItem,LagouItemLoader  #導入items容器類,和ItemLoader類
import time


class LagouSpider(CrawlSpider):                     #創建爬蟲類
    name = 'lagou'                                  #爬蟲名稱
    allowed_domains = ['www.luyin.org']             #起始域名
    start_urls = ['http://www.luyin.org/']          #起始url

    custom_settings = {
        "AUTOTHROTTLE_ENABLED": True,                             #覆蓋掉settings.py里的相同設置，開啟COOKIES
        "DOWNLOAD_DELAY":5
    }

    rules = (
        #配置抓取列表頁規則
        Rule(LinkExtractor(allow=('ggwa/.*')), follow=True),

        #配置抓取內容頁規則
        Rule(LinkExtractor(allow=('post/\d+.html.*')), callback='parse_job', follow=True),
    )

    def parse_job(self, response):                  #回調函數，注意：因為CrawlS模板的源碼創建了parse回調函數，所以切記我們不能創建parse名稱的函數
        atime = time.localtime(time.time())         #獲取系統當前時間
        dqatime = "{0}-{1}-{2} {3}:{4}:{5}".format(
            atime.tm_year,
            atime.tm_mon,
            atime.tm_mday,
            atime.tm_hour,
            atime.tm_min,
            atime.tm_sec
        )  # 將格式化時間日期，單獨取出來拼接成一個完整日期

        url = response.url

        item_loader = LagouItemLoader(LagouItem(), response=response)   # 將數據填充進items.py文件的LagouItem
        item_loader.add_xpath('title', '/html/head/title/text()') item_loader.add_xpath('description', '/html/head/meta[@name="Description"]/@content') item_loader.add_xpath('keywords', '/html/head/meta[@name="keywords"]/@content') item_loader.add_value('url', url) item_loader.add_value('riqi', dqatime) article_item = item_loader.load_item() yield article_item

 

items.py文件

# -*- coding: utf-8 -*-

# Define here the models for your scraped items
#
# See documentation in:
# http://doc.scrapy.org/en/latest/topics/items.html
#items.py,文件是專門用於，接收爬蟲獲取到的數據信息的，就相當於是容器文件

import scrapy
from scrapy.loader.processors import MapCompose,TakeFirst
from scrapy.loader import ItemLoader                #導入ItemLoader類也就加載items容器類填充數據
from adc.models.elasticsearch_orm import lagouType  #導入elasticsearch操作模塊

class LagouItemLoader(ItemLoader):                  #自定義Loader繼承ItemLoader類，在爬蟲頁面調用這個類填充數據到Item類
    default_output_processor = TakeFirst()          #默認利用ItemLoader類，加載items容器類填充數據，是列表類型，可以通過TakeFirst()方法，獲取到列表里的內容


def tianjia(value):                                 #自定義數據預處理函數
    return value                                    #將處理后的數據返給Item


class LagouItem(scrapy.Item):                       #設置爬蟲獲取到的信息容器類
    title = scrapy.Field(                           #接收爬蟲獲取到的title信息
        input_processor=MapCompose(tianjia),        #將數據預處理函數名稱傳入MapCompose方法里處理，數據預處理函數的形式參數value會自動接收字段title
 ) description = scrapy.Field() keywords = scrapy.Field() url = scrapy.Field() riqi = scrapy.Field() def save_to_es(self): lagou = lagouType() # 實例化elasticsearch(搜索引擎對象) lagou.title = self['title'] # 字段名稱=值 lagou.description = self['description'] lagou.keywords = self['keywords'] lagou.url = self['url'] lagou.riqi = self['riqi'] lagou.save() # 將數據寫入elasticsearch(搜索引擎對象) return

 

pipelines.py文件

# -*- coding: utf-8 -*-

# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: http://doc.scrapy.org/en/latest/topics/item-pipeline.html
from adc.models.elasticsearch_orm import lagouType  #導入elasticsearch操作模塊

class AdcPipeline(object):
    def process_item(self, item, spider):

        #也可以在這里將數據寫入elasticsearch搜索引擎，這里的缺點是統一處理
        # lagou = lagouType()
        # lagou.title = item['title']
        # lagou.description = item['description']
        # lagou.keywords = item['keywords']
        # lagou.url = item['url']
        # lagou.riqi = item['riqi']
        # lagou.save()
        item.save_to_es()       #執行items.py文件的save_to_es方法將數據寫入elasticsearch搜索引擎
        return item

 

settings.py文件，注冊pipelines

# Configure item pipelines
# See http://scrapy.readthedocs.org/en/latest/topics/item-pipeline.html
ITEM_PIPELINES = { 'adc.pipelines.AdcPipeline': 300, }

 

main.py爬蟲啟動文件

#!/usr/bin/env python
# -*- coding:utf8 -*-

from scrapy.cmdline import execute  #導入執行scrapy命令方法
import sys
import os

sys.path.append(os.path.join(os.getcwd())) #給Python解釋器，添加模塊新路徑 ,將main.py文件所在目錄添加到Python解釋器
 execute(['scrapy', 'crawl', 'lagou', '--nolog'])  #執行scrapy命令

# execute(['scrapy', 'crawl', 'lagou'])  #執行scrapy命令

 

運行爬蟲

 

寫入elasticsearch(搜索引擎)情況

 

 

補充：elasticsearch-dsl  的 增刪改查

#!/usr/bin/env python
# -*- coding:utf8 -*-
from datetime import datetime
from elasticsearch_dsl import DocType, Date, Nested, Boolean, \ analyzer, InnerObjectWrapper, Completion, Keyword, Text, Integer # 更多字段類型見第三百六十四節elasticsearch(搜索引擎)的mapping映射管理

from elasticsearch_dsl.connections import connections       # 導入連接elasticsearch(搜索引擎)服務器方法
connections.create_connection(hosts=['127.0.0.1']) class lagouType(DocType):                                                   # 自定義一個類來繼承DocType類
    # Text類型需要分詞，所以需要知道中文分詞器，ik_max_wordwei為中文分詞器
    title = Text(analyzer="ik_max_word")                                    # 設置，字段名稱=字段類型，Text為字符串類型並且可以分詞建立倒排索引
    description = Text(analyzer="ik_max_word") keywords = Text(analyzer="ik_max_word") url = Keyword()                                                         # 設置，字段名稱=字段類型，Keyword為普通字符串類型，不分詞
    riqi = Date()                                                           # 設置，字段名稱=字段類型，Date日期類型

    class Meta:                                                             # Meta是固定寫法
        index = "lagou"                                                     # 設置索引名稱(相當於數據庫名稱)
        doc_type = 'biao'                                                   # 設置表名稱

if __name__ == "__main__":          # 判斷在本代碼文件執行才執行里面的方法，其他頁面調用的則不執行里面的方法
    lagouType.init()                # 生成elasticsearch(搜索引擎)的索引，表，字段等信息


# 使用方法說明：
# 在要要操作elasticsearch(搜索引擎)的頁面，導入此模塊
# lagou = lagouType()           #實例化類
# lagou.title = '值'            #要寫入字段=值
# lagou.description = '值'
# lagou.keywords = '值'
# lagou.url = '值'
# lagou.riqi = '值'
# lagou.save()                  #將數據寫入elasticsearch(搜索引擎)

 

1新增數據

from adc.models.elasticsearch_orm import lagouType  #導入剛才配置的elasticsearch操作模塊

　　　　　lagou = lagouType()                         # 實例化elasticsearch(搜索引擎對象)
　　　　　lagou._id = 1　　　　　　　　　　　　　#自定義ID，很重要，以后都是根據ID來操作

        lagou.title = self['title']                 # 字段名稱=值
        lagou.description = self['description']
        lagou.keywords = self['keywords']
        lagou.url = self['url']
        lagou.riqi = self['riqi']
        lagou.save()                                # 將數據寫入elasticsearch(搜索引擎對象)

 

2刪除指定數據

from adc.models.elasticsearch_orm import lagouType  #導入剛才配置的elasticsearch操作模塊

sousuo_orm = lagouType()                    # 實例化
sousuo_orm.get(id=1).delete()               # 刪除id等於1的數據

 

3修改指定的數據

from adc.models.elasticsearch_orm import lagouType  #導入剛才配置的elasticsearch操作模塊

sousuo_orm = lagouType()                           # 實例化
sousuo_orm.get(id=1).update(title='123456789')     # 修改id等於1的數據

以上全部使用elasticsearch-dsl模塊

 

 

注意下面使用的原生elasticsearch模塊

刪除指定使用，就是相當於刪除指定數據庫

使用原生elasticsearch模塊刪除指定索引

from elasticsearch import Elasticsearch                                     # 導入原生的elasticsearch(搜索引擎)接口
client = Elasticsearch(hosts=settings.Elasticsearch_hosts)                  # 連接原生的elasticsearch

# 使用原生elasticsearch模塊刪除指定索引
#要做容錯處理，如果索引不存在會報錯
            try:
                client.indices.delete(index='jxiou_zuopin') except Exception as e:
                pass

原生查詢

from elasticsearch import Elasticsearch                 # 導入原生的elasticsearch(搜索引擎)接口
            client = Elasticsearch(hosts=Elasticsearch_hosts)       # 連接原生的elasticsearch


response = client.search(                               # 原生的elasticsearch接口的search()方法，就是搜索，可以支持原生elasticsearch語句查詢
                index="jxiou_zuopin",                               # 設置索引名稱
                doc_type="zuopin",                                  # 設置表名稱
                body={                                              # 書寫elasticsearch語句
                    "query": {
                        "multi_match": {                            # multi_match查詢
                            "query": sousuoci,                      # 查詢關鍵詞
                            "fields": ["title"]                     # 查詢字段
                        }
                    },
                    "from": (page - 1) * tiaoshu,                   # 從第幾條開始獲取
                    "size": tiaoshu,                                # 獲取多少條數據
                    "highlight": {                                  # 查詢關鍵詞高亮處理
                        "pre_tags": ['<span class="gaoliang">'],    # 高亮開始標簽
                        "post_tags": ['</span>'],                   # 高亮結束標簽
                        "fields": {                                 # 高亮設置
                            "title": {}                             # 高亮字段
                        }
                    }
                }
            )
            # 開始獲取數據
            total_nums = response["hits"]["total"]                  # 獲取查詢結果的總條數

            hit_list = []                                           # 設置一個列表來儲存搜索到的信息，返回給html頁面


            for hit in response["hits"]["hits"]:                                # 循環查詢到的結果
                hit_dict = {}                                                   # 設置一個字典來儲存循環結果
                if "title" in hit["highlight"]:                                 # 判斷title字段，如果高亮字段有類容
                    hit_dict["title"] = "".join(hit["highlight"]["title"])      # 獲取高亮里的title
                else:
                    hit_dict["title"] = hit["_source"]["title"]                 # 否則獲取不是高亮里的title

                hit_dict["id"] = hit["_source"]["nid"]                          # 獲取返回nid

                # 加密樣音地址
                hit_dict["yangsrc"] = jia_mi(str(hit["_source"]["yangsrc"]))    # 獲取返回yangsrc

                hit_list.append(hit_dict)