ElasticSearch python基本操作

待我称王封你为后i 2021-12-16 12:41 417阅读 0赞

## 　官方文档：[https://elasticsearch-py.readthedocs.io/en/master/][https_elasticsearch-py.readthedocs.io_en_master] ##

## 　　1、介绍 ##

python提供了操作ElasticSearch 接口，因此要用python来操作ElasticSearch，首先要安装python的ElasticSearch包，用命令 pip install elasticsearch安装或下载安装：[https://pypi.python.org/pypi/elasticsearch/5.4.0][https_pypi.python.org_pypi_elasticsearch_5.4.0]

## 　　2、创建索引 ##

假如创建索引名称为ott，类型为ott\_type的索引，该索引中有五个字段：

title：存储中文标题，

date：存储日期格式（2017-09-08），

keyword：存储中文关键字，

source：存储中文来源，

link：存储链接，

创建映射：

![194720-20170925154308120-640835864.png][]

![194720-20170925155522932-1432752584.png][]

## 　　3、索引数据 ##

![194720-20170925154453401-1686365374.png][]

批量索引

利用bulk批量索引数据

![194720-20170925154613932-594672450.png][]

## 　　4、查询索引 ##

![194720-20170925155348792-770510036.png][]

## 　　5、删除数据 ##

![194720-20170925154658401-155609552.png][]

##  　　6、完整代码 ##

#coding:utf8
    import os
    import time
    from os import walk
    import CSVOP
    from datetime import datetime
    from elasticsearch import Elasticsearch
    from elasticsearch.helpers import bulk
    
    class ElasticObj:
        def __init__(self, index_name,index_type,ip ="127.0.0.1"):
            '''
    
            :param index_name: 索引名称
            :param index_type: 索引类型
            '''
            self.index_name =index_name
            self.index_type = index_type
            # 无用户名密码状态
            #self.es = Elasticsearch([ip])
            #用户名密码状态
            self.es = Elasticsearch([ip],http_auth=('elastic', 'password'),port=9200)
    
        def create_index(self,index_name="ott",index_type="ott_type"):
            '''
            创建索引,创建索引名称为ott，类型为ott_type的索引
            :param ex: Elasticsearch对象
            :return:
            '''
            #创建映射
            _index_mappings = {
                "mappings": {
                    self.index_type: {
                        "properties": {
                            "title": {
                                "type": "text",
                                "index": True,
                                "analyzer": "ik_max_word",
                                "search_analyzer": "ik_max_word"
                            },
                            "date": {
                                "type": "text",
                                "index": True
                            },
                            "keyword": {
                                "type": "string",
                                "index": "not_analyzed"
                            },
                            "source": {
                                "type": "string",
                                "index": "not_analyzed"
                            },
                            "link": {
                                "type": "string",
                                "index": "not_analyzed"
                            }
                        }
                    }
    
                }
            }
            if self.es.indices.exists(index=self.index_name) is not True:
                res = self.es.indices.create(index=self.index_name, body=_index_mappings)
                print res
    
    
        def IndexData(self):
            es = Elasticsearch()
            csvdir = 'D:/work/ElasticSearch/exportExcels'
            filenamelist = []
            for (dirpath, dirnames, filenames) in walk(csvdir):
                filenamelist.extend(filenames)
                break
            total = 0
            for file in filenamelist:
                csvfile = csvdir + '/' + file
                self.Index_Data_FromCSV(csvfile,es)
                total += 1
                print total
                time.sleep(10)
    
        def Index_Data_FromCSV(self,csvfile):
            '''
            从CSV文件中读取数据，并存储到es中
            :param csvfile: csv文件，包括完整路径
            :return:
            '''
            list = CSVOP.ReadCSV(csvfile)
            index = 0
            doc = {}
            for item in list:
                if index > 1:#第一行是标题
                    doc['title'] = item[0]
                    doc['link'] = item[1]
                    doc['date'] = item[2]
                    doc['source'] = item[3]
                    doc['keyword'] = item[4]
                    res = self.es.index(index=self.index_name, doc_type=self.index_type, body=doc)
                    print(res['created'])
                index += 1
                print index
    
        def Index_Data(self):
            '''
            数据存储到es
            :return:
            '''
            list = [
                {   "date": "2017-09-13",
                    "source": "慧聪网",
                    "link": "http://info.broadcast.hc360.com/2017/09/130859749974.shtml",
                    "keyword": "电视",
                    "title": "付费 电视 行业面临的转型和挑战"
                 },
                {   "date": "2017-09-13",
                    "source": "中国文明网",
                    "link": "http://www.wenming.cn/xj_pd/yw/201709/t20170913_4421323.shtml",
                    "keyword": "电视",
                    "title": "电视 专题片《巡视利剑》广获好评：铁腕反腐凝聚党心民心"
                 }
                  ]
            for item in list:
                res = self.es.index(index=self.index_name, doc_type=self.index_type, body=item)
                print(res['created'])
    
        def bulk_Index_Data(self):
            '''
            用bulk将批量数据存储到es
            :return:
            '''
            list = [
                {"date": "2017-09-13",
                 "source": "慧聪网",
                 "link": "http://info.broadcast.hc360.com/2017/09/130859749974.shtml",
                 "keyword": "电视",
                 "title": "付费 电视 行业面临的转型和挑战"
                 },
                {"date": "2017-09-13",
                 "source": "中国文明网",
                 "link": "http://www.wenming.cn/xj_pd/yw/201709/t20170913_4421323.shtml",
                 "keyword": "电视",
                 "title": "电视 专题片《巡视利剑》广获好评：铁腕反腐凝聚党心民心"
                 },
                {"date": "2017-09-13",
                 "source": "人民电视",
                 "link": "http://tv.people.com.cn/BIG5/n1/2017/0913/c67816-29533981.html",
                 "keyword": "电视",
                 "title": "中国第21批赴刚果（金）维和部隊启程--人民 电视 --人民网"
                 },
                {"date": "2017-09-13",
                 "source": "站长之家",
                 "link": "http://www.chinaz.com/news/2017/0913/804263.shtml",
                 "keyword": "电视",
                 "title": "电视 盒子 哪个牌子好？ 吐血奉献三大选购秘笈"
                 }
            ]
            ACTIONS = []
            i = 1
            for line in list:
                action = {
                    "_index": self.index_name,
                    "_type": self.index_type,
                    "_id": i, #_id 也可以默认生成，不赋值
                    "_source": {
                        "date": line['date'],
                        "source": line['source'].decode('utf8'),
                        "link": line['link'],
                        "keyword": line['keyword'].decode('utf8'),
                        "title": line['title'].decode('utf8')}
                }
                i += 1
                ACTIONS.append(action)
                # 批量处理
            success, _ = bulk(self.es, ACTIONS, index=self.index_name, raise_on_error=True)
            print('Performed %d actions' % success)
    
        def Delete_Index_Data(self,id):
            '''
            删除索引中的一条
            :param id:
            :return:
            '''
            res = self.es.delete(index=self.index_name, doc_type=self.index_type, id=id)
            print res
    
        def Get_Data_Id(self,id):
    
            res = self.es.get(index=self.index_name, doc_type=self.index_type,id=id)
            print(res['_source'])
    
            print '------------------------------------------------------------------'
            #
            # # 输出查询到的结果
            for hit in res['hits']['hits']:
                # print hit['_source']
                print hit['_source']['date'],hit['_source']['source'],hit['_source']['link'],hit['_source']['keyword'],hit['_source']['title']
    
        def Get_Data_By_Body(self):
            # doc = {'query': {'match_all': {}}}
            doc = {
                "query": {
                    "match": {
                        "keyword": "电视"
                    }
                }
            }
            _searched = self.es.search(index=self.index_name, doc_type=self.index_type, body=doc)
    
            for hit in _searched['hits']['hits']:
                # print hit['_source']
                print hit['_source']['date'], hit['_source']['source'], hit['_source']['link'], hit['_source']['keyword'], \
                hit['_source']['title']
    
    
    
    
    obj =ElasticObj("ott","ott_type",ip ="47.93.117.127")
    # obj = ElasticObj("ott1", "ott_type1")
    
    # obj.create_index()
    obj.Index_Data()
    # obj.bulk_Index_Data()
    # obj.IndexData()
    # obj.Delete_Index_Data(1)
    # csvfile = 'D:/work/ElasticSearch/exportExcels/2017-08-31_info.csv'
    # obj.Index_Data_FromCSV(csvfile)
    # obj.GetData(es)

转自：[https://www.cnblogs.com/shaosks/p/7592229.html][https_www.cnblogs.com_shaosks_p_7592229.html]

[https_elasticsearch-py.readthedocs.io_en_master]: https://elasticsearch-py.readthedocs.io/en/master/
[https_pypi.python.org_pypi_elasticsearch_5.4.0]: https://pypi.python.org/pypi/elasticsearch/5.4.0
[194720-20170925154308120-640835864.png]: /images/20211214/e9db7562c789472ebbeef215135b4b7b.png
[194720-20170925155522932-1432752584.png]: /images/20211214/e185b7bfffee43debf5b60d9455432ef.png
[194720-20170925154453401-1686365374.png]: /images/20211214/5c6386c3509440d49baeb7c430081cc5.png
[194720-20170925154613932-594672450.png]: /images/20211214/dc4ecd75f6414fada7e3b34be249ed84.png
[194720-20170925155348792-770510036.png]: /images/20211214/979e38b7a42e46f0a07dcf9f1781147f.png
[194720-20170925154658401-155609552.png]: /images/20211214/1a548a8bba4d4da18f7f3b7ffb37cdac.png
[https_www.cnblogs.com_shaosks_p_7592229.html]: https://www.cnblogs.com/shaosks/p/7592229.html