手把手教你使用Flask搭建ES搜索引擎(预备篇)

[[406279]]

1 前言

Elasticsearch 是一个开源的搜索引擎，建立在一个全文搜索引擎库 Apache Lucene™ 基础之上。

10年积累的成都网站建设、网站制作经验，可以快速应对客户对网站的新想法和需求。提供各种问题对应的解决方案。让选择我们的客户得到更好、更有力的网络服务。我虽然不认识你，你也不认识我。但先网站设计后付款的网站建设流程，更有新建免费网站建设让你可以放心的选择与我们合作。

那么如何实现 Elasticsearch和 Python 的对接成为我们所关心的问题了 (怎么什么都要和 Python 关联啊)。

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.

默认端口 9200，初始化前请确保本地已搭建好 Elasticsearch的所属环境。

根据 ID 获取文档数据

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

插入文档数据

 
 
 
 
    
  
  
  
  
  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.

搜索文档数据

 
 
 
 
    
  
  
  
  
  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: 返回的数据大小 
    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.

删除文档数据

 
 
 
 
    
  
  
  
  
  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 类也是为了方便调用，整体贴一下。

 
 
 
 
    
  
  
  
  
  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 中的数据插入到 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/

发现新加的 spider 数据文档确实已经进去了。

3 爬虫入库

要想实现 ES 搜索，首先要有数据支持，而海量的数据往往来自爬虫。

为了节省时间，编写一个最简单的爬虫，抓取 百度百科。

简单粗暴一点，先 递归获取 很多很多的 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. # 当爬取深度小于10层时，递归调用主函数，继续爬取第二层的所有链接 
    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 文件中之后，然后启动任务。

 
 
 
 
    
  
  
  
  
  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("//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 任务队列，gevent 模式，-c 就是10个线程刷刷刷就干起来了，速度杠杠的 !!

啥?分布式? 那就加多几台机器啦，直接把代码拷贝到目标服务器，通过 redis 共享队列协同多机抓取。

这里是先将数据存储到了 MongoDB 上(个人习惯)，你也可以直接存到 ES 中，但是单条单条的插入速度堪忧(接下来会讲到优化，哈哈)。

使用前面的例子将 Mongo 中的数据批量导入到 ES 中，OK !!!

到这一个简单的数据抓取就已经完毕了。

好啦，现在 ES 中已经有了数据啦，接下来就应该是 Flask web 的操作啦，当然，Django，FastAPI 也很优秀。嘿嘿，你喜欢 !!

网页名称：手把手教你使用Flask搭建ES搜索引擎(预备篇)
标题URL：http://www.shufengxianlan.com/qtweb/news6/148706.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

建站公司知识

各行业网站

成都微信小程序　　 　重庆app软件开发　　 　深圳惠宜防蚊纱窗　　 　成都做网站　　 　做网站　　 　成都户外广告牌设计　　 　工业净化空调　　 　成都定制网站　　 　成都网站推广　　 　绵阳网站制作公司　　 　成都网站建设　　 　温江网站建设　　 　主机租用　　 　物流公司网站建设方案　　 　东电技服　　 　成都石膏隔墙板　　 　发电机组维修保养　　 　宜宾做网站　　 　整合营销　　 　网络推广外包