日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時間：8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案
手把手教你使用Flask搭建ES搜索引擎(預(yù)備篇)

[[406279]]

1 前言

Elasticsearch 是一個開源的搜索引擎，建立在一個全文搜索引擎庫 Apache Lucene? 基礎(chǔ)之上。

10年積累的成都網(wǎng)站建設(shè)、網(wǎng)站制作經(jīng)驗，可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你，你也不認(rèn)識我。但先網(wǎng)站設(shè)計后付款的網(wǎng)站建設(shè)流程，更有新建免費網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。

那么如何實現(xiàn) Elasticsearch和 Python 的對接成為我們所關(guān)心的問題了 (怎么什么都要和 Python 關(guān)聯(lián)啊)。

2 Python 交互

所以，Python 也就提供了可以對接 Elasticsearch的依賴庫。

 
 
 
 
    
  
  
  
  
  1. pip install elasticsearch 
    2.

初始化連接一個 Elasticsearch 操作對象。

 
 
 
 
    
  
  
  
  
  1. def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"): 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3.     # self.es = Elasticsearch([ip], http_auth=('username', 'password'), port=9200) 
    4. 
  
  
  
  
  4.     self.es = Elasticsearch("localhost:9200") 
    5. 
  
  
  
  
  5.     self.index_type = index_type 
    6. 
  
  
  
  
  6.     self.index_name = index_name 
    7.

默認(rèn)端口 9200，初始化前請確保本地已搭建好 Elasticsearch的所屬環(huán)境。

根據(jù) ID 獲取文檔數(shù)據(jù)

 
 
 
 
    
  
  
  
  
  1. def get_doc(self, uid): 
    2. 
  
  
  
  
  2.     return self.es.get(index=self.index_name, id=uid) 
    3.

插入文檔數(shù)據(jù)

 
 
 
 
    
  
  
  
  
  1. def insert_one(self, doc: dict): 
    2. 
  
  
  
  
  2.     self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4. def insert_array(self, docs: list): 
    5. 
  
  
  
  
  5.     for doc in docs: 
    6. 
  
  
  
  
  6.         self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
    7.

搜索文檔數(shù)據(jù)

 
 
 
 
    
  
  
  
  
  1. def search(self, query, count: int = 30): 
    2. 
  
  
  
  
  2.     dsl = { 
    3. 
  
  
  
  
  3.         "query": { 
    4. 
  
  
  
  
  4.             "multi_match": { 
    5. 
  
  
  
  
  5.                 "query": query, 
    6. 
  
  
  
  
  6.                 "fields": ["title", "content", "link"] 
    7. 
  
  
  
  
  7.             } 
    8. 
  
  
  
  
  8.         }, 
    9. 
  
  
  
  
  9.         "highlight": { 
    10. 
  
  
  
  
  10.             "fields": { 
    11. 
  
  
  
  
  11.                 "title": {} 
    12. 
  
  
  
  
  12.             } 
    13. 
  
  
  
  
  13.         } 
    14. 
  
  
  
  
  14.     } 
    15. 
  
  
  
  
  15.     match_data = self.es.search(index=self.index_name, body=dsl, size=count) 
    16. 
  
  
  
  
  16.     return match_data 
    17. 
  
  
  
  
  17.  
    18. 
  
  
  
  
  18. def __search(self, query: dict, count: int = 20): # count: 返回的數(shù)據(jù)大小 
    19. 
  
  
  
  
  19.     results = [] 
    20. 
  
  
  
  
  20.     params = { 
    21. 
  
  
  
  
  21.         'size': count 
    22. 
  
  
  
  
  22.     } 
    23. 
  
  
  
  
  23.     match_data = self.es.search(index=self.index_name, body=query, params=params) 
    24. 
  
  
  
  
  24.     for hit in match_data['hits']['hits']: 
    25. 
  
  
  
  
  25.         results.append(hit['_source']) 
    26. 
  
  
  
  
  26.  
    27. 
  
  
  
  
  27.     return results 
    28.

刪除文檔數(shù)據(jù)

 
 
 
 
    
  
  
  
  
  1. def delete_index(self): 
    2. 
  
  
  
  
  2.     try: 
    3. 
  
  
  
  
  3.         self.es.indices.delete(index=self.index_name) 
    4. 
  
  
  
  
  4.     except: 
    5. 
  
  
  
  
  5.         pass 
    6.

好啊，封裝 search 類也是為了方便調(diào)用，整體貼一下。

 
 
 
 
    
  
  
  
  
  1. from elasticsearch import Elasticsearch 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3.  
    4. 
  
  
  
  
  4. class elasticSearch(): 
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6.     def __init__(self, index_type: str, index_name: str, ip="127.0.0.1"): 
    7. 
  
  
  
  
  7.  
    8. 
  
  
  
  
  8.         # self.es = Elasticsearch([ip], http_auth=('elastic', 'password'), port=9200) 
    9. 
  
  
  
  
  9.         self.es = Elasticsearch("localhost:9200") 
    10. 
  
  
  
  
  10.         self.index_type = index_type 
    11. 
  
  
  
  
  11.         self.index_name = index_name 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13.     def create_index(self): 
    14. 
  
  
  
  
  14.         if self.es.indices.exists(index=self.index_name) is True: 
    15. 
  
  
  
  
  15.             self.es.indices.delete(index=self.index_name) 
    16. 
  
  
  
  
  16.         self.es.indices.create(index=self.index_name, ignore=400) 
    17. 
  
  
  
  
  17.  
    18. 
  
  
  
  
  18.     def delete_index(self): 
    19. 
  
  
  
  
  19.         try: 
    20. 
  
  
  
  
  20.             self.es.indices.delete(index=self.index_name) 
    21. 
  
  
  
  
  21.         except: 
    22. 
  
  
  
  
  22.             pass 
    23. 
  
  
  
  
  23.  
    24. 
  
  
  
  
  24.     def get_doc(self, uid): 
    25. 
  
  
  
  
  25.         return self.es.get(index=self.index_name, id=uid) 
    26. 
  
  
  
  
  26.  
    27. 
  
  
  
  
  27.     def insert_one(self, doc: dict): 
    28. 
  
  
  
  
  28.         self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
    29. 
  
  
  
  
  29.  
    30. 
  
  
  
  
  30.     def insert_array(self, docs: list): 
    31. 
  
  
  
  
  31.         for doc in docs: 
    32. 
  
  
  
  
  32.             self.es.index(index=self.index_name, doc_type=self.index_type, body=doc) 
    33. 
  
  
  
  
  33.  
    34. 
  
  
  
  
  34.     def search(self, query, count: int = 30): 
    35. 
  
  
  
  
  35.         dsl = { 
    36. 
  
  
  
  
  36.             "query": { 
    37. 
  
  
  
  
  37.                 "multi_match": { 
    38. 
  
  
  
  
  38.                     "query": query, 
    39. 
  
  
  
  
  39.                     "fields": ["title", "content", "link"] 
    40. 
  
  
  
  
  40.                 } 
    41. 
  
  
  
  
  41.             }, 
    42. 
  
  
  
  
  42.             "highlight": { 
    43. 
  
  
  
  
  43.                 "fields": { 
    44. 
  
  
  
  
  44.                     "title": {} 
    45. 
  
  
  
  
  45.                 } 
    46. 
  
  
  
  
  46.             } 
    47. 
  
  
  
  
  47.         } 
    48. 
  
  
  
  
  48.         match_data = self.es.search(index=self.index_name, body=dsl, size=count) 
    49. 
  
  
  
  
  49.         return match_data 
    50.

嘗試一下把 Mongodb 中的數(shù)據(jù)插入到 ES 中。

 
 
 
 
    
  
  
  
  
  1. import json 
    2. 
  
  
  
  
  2. from datetime import datetime 
    3. 
  
  
  
  
  3. import pymongo 
    4. 
  
  
  
  
  4. from app.elasticsearchClass import elasticSearch 
    5. 
  
  
  
  
  5.  
    6. 
  
  
  
  
  6. client = pymongo.MongoClient('127.0.0.1', 27017) 
    7. 
  
  
  
  
  7. db = client['spider'] 
    8. 
  
  
  
  
  8. sheet = db.get_collection('Spider').find({}, {'_id': 0, }) 
    9. 
  
  
  
  
  9.  
    10. 
  
  
  
  
  10. es = elasticSearch(index_type="spider_data",index_name="spider") 
    11. 
  
  
  
  
  11. es.create_index() 
    12. 
  
  
  
  
  12.  
    13. 
  
  
  
  
  13. for i in sheet: 
    14. 
  
  
  
  
  14.     data = { 
    15. 
  
  
  
  
  15.             'title': i["title"], 
    16. 
  
  
  
  
  16.             'content':i["data"], 
    17. 
  
  
  
  
  17.             'link': i["link"], 
    18. 
  
  
  
  
  18.             'create_time':datetime.now() 
    19. 
  
  
  
  
  19.         } 
    20. 
  
  
  
  
  20.  
    21. 
  
  
  
  
  21.     es.insert_one(doc=data) 
    22.

到 ES 中查看一下，啟動 elasticsearch-head 插件。

如果是 npm 安裝的那么 cd 到根目錄之后直接 npm run start 就跑起來了。

本地訪問 http://localhost:9100/

發(fā)現(xiàn)新加的 spider 數(shù)據(jù)文檔確實已經(jīng)進去了。

3 爬蟲入庫

要想實現(xiàn) ES 搜索，首先要有數(shù)據(jù)支持，而海量的數(shù)據(jù)往往來自爬蟲。

為了節(jié)省時間，編寫一個最簡單的爬蟲，抓取 百度百科。

簡單粗暴一點，先 遞歸獲取 很多很多的 url 鏈接

 
 
 
 
    
  
  
  
  
  1. import requests 
    2. 
  
  
  
  
  2. import re 
    3. 
  
  
  
  
  3. import time 
    4. 
  
  
  
  
  4.  
    5. 
  
  
  
  
  5. exist_urls = [] 
    6. 
  
  
  
  
  6. headers = { 
    7. 
  
  
  
  
  7.     'User-Agent': 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.62 Safari/537.36', 
    8. 
  
  
  
  
  8. } 
    9. 
  
  
  
  
  9.  
    10. 
  
  
  
  
  10. def get_link(url): 
    11. 
  
  
  
  
  11.     try: 
    12. 
  
  
  
  
  12.         response = requests.get(url=url, headers=headers) 
    13. 
  
  
  
  
  13.         response.encoding = 'UTF-8' 
    14. 
  
  
  
  
  14.         html = response.text 
    15. 
  
  
  
  
  15.         link_lists = re.findall('.*?]*?)".*?', html) 
    16. 
  
  
  
  
  16.         return link_lists 
    17. 
  
  
  
  
  17.     except Exception as e: 
    18. 
  
  
  
  
  18.         pass 
    19. 
  
  
  
  
  19.     finally: 
    20. 
  
  
  
  
  20.         exist_urls.append(url) 
    21. 
  
  
  
  
  21.  
    22. 
  
  
  
  
  22.  
    23. 
  
  
  
  
  23. # 當(dāng)爬取深度小于10層時，遞歸調(diào)用主函數(shù)，繼續(xù)爬取第二層的所有鏈接 
    24. 
  
  
  
  
  24. def main(start_url, depth=1): 
    25. 
  
  
  
  
  25.     link_lists = get_link(start_url) 
    26. 
  
  
  
  
  26.     if link_lists: 
    27. 
  
  
  
  
  27.         unique_lists = list(set(link_lists) - set(exist_urls)) 
    28. 
  
  
  
  
  28.         for unique_url in unique_lists: 
    29. 
  
  
  
  
  29.             unique_url = 'https://baike.baidu.com/item/' + unique_url 
    30. 
  
  
  
  
  30.  
    31. 
  
  
  
  
  31.             with open('url.txt', 'a+') as f: 
    32. 
  
  
  
  
  32.                 f.write(unique_url + '\n') 
    33. 
  
  
  
  
  33.                 f.close() 
    34. 
  
  
  
  
  34.         if depth < 10: 
    35. 
  
  
  
  
  35.             main(unique_url, depth + 1) 
    36. 
  
  
  
  
  36.  
    37. 
  
  
  
  
  37. if __name__ == '__main__': 
    38. 
  
  
  
  
  38.     start_url = 'https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91' 
    39. 
  
  
  
  
  39.     main(start_url) 
    40.

把全部 url 存到 url.txt 文件中之后，然后啟動任務(wù)。

 
 
 
 
    
  
  
  
  
  1. # parse.py 
    2. 
  
  
  
  
  2. from celery import Celery 
    3. 
  
  
  
  
  3. import requests 
    4. 
  
  
  
  
  4. from lxml import etree 
    5. 
  
  
  
  
  5. import pymongo 
    6. 
  
  
  
  
  6. app = Celery('tasks', broker='redis://localhost:6379/2') 
    7. 
  
  
  
  
  7. client = pymongo.MongoClient('localhost',27017) 
    8. 
  
  
  
  
  8. db = client['baike'] 
    9. 
  
  
  
  
  9. @app.task 
    10. 
  
  
  
  
  10. def get_url(link): 
    11. 
  
  
  
  
  11.     item = {} 
    12. 
  
  
  
  
  12.     headers = {'User-Agent':'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/34.0.1847.131 Safari/537.36'} 
    13. 
  
  
  
  
  13.     res = requests.get(link,headers=headers) 
    14. 
  
  
  
  
  14.     res.encoding = 'UTF-8' 
    15. 
  
  
  
  
  15.     doc = etree.HTML(res.text) 
    16. 
  
  
  
  
  16.     content = doc.xpath("http://div[@class='lemma-summary']/div[@class='para']//text()") 
    17. 
  
  
  
  
  17.     print(res.status_code) 
    18. 
  
  
  
  
  18.     print(link,'\t','++++++++++++++++++++') 
    19. 
  
  
  
  
  19.     item['link'] = link 
    20. 
  
  
  
  
  20.     data = ''.join(content).replace(' ', '').replace('\t', '').replace('\n', '').replace('\r', '') 
    21. 
  
  
  
  
  21.     item['data'] = data 
    22. 
  
  
  
  
  22.     if db['Baike'].insert(dict(item)): 
    23. 
  
  
  
  
  23.         print("is OK ...") 
    24. 
  
  
  
  
  24.     else: 
    25. 
  
  
  
  
  25.         print('Fail') 
    26.

run.py 飛起來

 
 
 
 
    
  
  
  
  
  1. from parse import get_url 
    2. 
  
  
  
  
  2.  
    3. 
  
  
  
  
  3. def main(url): 
    4. 
  
  
  
  
  4.     result = get_url.delay(url) 
    5. 
  
  
  
  
  5.     return result 
    6. 
  
  
  
  
  6.  
    7. 
  
  
  
  
  7. def run(): 
    8. 
  
  
  
  
  8.     with open('./url.txt', 'r') as f: 
    9. 
  
  
  
  
  9.         for url in f.readlines(): 
    10. 
  
  
  
  
  10.             main(url.strip('\n')) 
    11. 
  
  
  
  
  11.  
    12. 
  
  
  
  
  12. if __name__ == '__main__': 
    13. 
  
  
  
  
  13.     run() 
    14.

黑窗口鍵入

 
 
 
 
    
  
  
  
  
  1. celery -A parse worker -l info -P gevent -c 10 
    2.

哦豁 !! 你居然使用了 Celery 任務(wù)隊列，gevent 模式，-c 就是10個線程刷刷刷就干起來了，速度杠杠的 !!

啥?分布式? 那就加多幾臺機器啦，直接把代碼拷貝到目標(biāo)服務(wù)器，通過 redis 共享隊列協(xié)同多機抓取。

這里是先將數(shù)據(jù)存儲到了 MongoDB 上(個人習(xí)慣)，你也可以直接存到 ES 中，但是單條單條的插入速度堪憂(接下來會講到優(yōu)化，哈哈)。

使用前面的例子將 Mongo 中的數(shù)據(jù)批量導(dǎo)入到 ES 中，OK !!!

到這一個簡單的數(shù)據(jù)抓取就已經(jīng)完畢了。

好啦，現(xiàn)在 ES 中已經(jīng)有了數(shù)據(jù)啦，接下來就應(yīng)該是 Flask web 的操作啦，當(dāng)然，Django，F(xiàn)astAPI 也很優(yōu)秀。嘿嘿，你喜歡 !!


分享標(biāo)題：手把手教你使用Flask搭建ES搜索引擎(預(yù)備篇)
文章地址：http://www.dlmjj.cn/article/dpsjcpc.html