文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

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

2024-12-03 03:33

关注

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

2 Python 交互

所以,Python 也就提供了可以对接 Elasticsearch的依赖库。

  1. pip install elasticsearch 

初始化连接一个 Elasticsearch 操作对象。

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

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

根据 ID 获取文档数据

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

插入文档数据

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

搜索文档数据

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

删除文档数据

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

好啊,封装 search 类也是为了方便调用,整体贴一下。

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

尝试一下把 Mongodb 中的数据插入到 ES 中。

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

到 ES 中查看一下,启动 elasticsearch-head 插件。

如果是 npm 安装的那么 cd 到根目录之后直接 npm run start 就跑起来了。

本地访问 http://localhost:9100/

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

3 爬虫入库

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

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

简单粗暴一点,先 递归获取 很多很多的 url 链接

  1. import requests 
  2. import re 
  3. import time 
  4.  
  5. exist_urls = [] 
  6. headers = { 
  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.  
  9. def get_link(url): 
  10.     try: 
  11.         response = requests.get(url=url, headers=headers) 
  12.         response.encoding = 'UTF-8' 
  13.         html = response.text 
  14.         link_lists = re.findall('.*?]*?)".*?', html) 
  15.         return link_lists 
  16.     except Exception as e: 
  17.         pass 
  18.     finally: 
  19.         exist_urls.append(url) 
  20.  
  21.  
  22. # 当爬取深度小于10层时,递归调用主函数,继续爬取第二层的所有链接 
  23. def main(start_url, depth=1): 
  24.     link_lists = get_link(start_url) 
  25.     if link_lists: 
  26.         unique_lists = list(set(link_lists) - set(exist_urls)) 
  27.         for unique_url in unique_lists: 
  28.             unique_url = 'https://baike.baidu.com/item/' + unique_url 
  29.  
  30.             with open('url.txt''a+'as f: 
  31.                 f.write(unique_url + '\n'
  32.                 f.close() 
  33.         if depth < 10: 
  34.             main(unique_url, depth + 1) 
  35.  
  36. if __name__ == '__main__'
  37.     start_url = 'https://baike.baidu.com/item/%E7%99%BE%E5%BA%A6%E7%99%BE%E7%A7%91' 
  38.     main(start_url) 

把全部 url 存到 url.txt 文件中之后,然后启动任务。

  1. # parse.py 
  2. from celery import Celery 
  3. import requests 
  4. from lxml import etree 
  5. import pymongo 
  6. app = Celery('tasks', broker='redis://localhost:6379/2'
  7. client = pymongo.MongoClient('localhost',27017) 
  8. db = client['baike'
  9. @app.task 
  10. def get_url(link): 
  11.     item = {} 
  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.     res = requests.get(link,headers=headers) 
  14.     res.encoding = 'UTF-8' 
  15.     doc = etree.HTML(res.text) 
  16.     content = doc.xpath("//div[@class='lemma-summary']/div[@class='para']//text()"
  17.     print(res.status_code) 
  18.     print(link,'\t','++++++++++++++++++++'
  19.     item['link'] = link 
  20.     data = ''.join(content).replace(' ''').replace('\t''').replace('\n''').replace('\r'''
  21.     item['data'] = data 
  22.     if db['Baike'].insert(dict(item)): 
  23.         print("is OK ..."
  24.     else
  25.         print('Fail'

run.py 飞起来

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

黑窗口键入

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

哦豁 !! 你居然使用了 Celery 任务队列,gevent 模式,-c 就是10个线程刷刷刷就干起来了,速度杠杠的 !!

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

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

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

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

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

 

来源:Python爬虫与数据挖掘 内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯