一、安装

pip install elasticsearch 

二、一个小封装类

#索引类
class ElasticSearchClient(object):
    # TODO：实例和事务化单个node，若需要多个node，需要重构代码
    def __init__(self, filepath="app/conf/conf.ini"):
        #读取es配置
        conf=configparser.ConfigParser()
        conf.read(filepath,encoding='utf-8')
        # TODO：传参

        self.es_servers = [{
            "host": conf.get('Elasticsearch','url'),
            "port": conf.get('Elasticsearch','port')
        }]
    # http_auth是对设置了安全机制的es库需要写入 账号与密码，如果没有设置则不用写这个参数
        self.es_client = elasticsearch.Elasticsearch(hosts=self.es_servers,http_auth=("xxx", "xxxxx")) 

    # TODO:进行创建一个数据库，即index
    def create_index(self, index_name):
        self.es_client.indices.create(index=index_name)
    # TODO：指定map创建一个数据库
    def createindex_by_map(self,index_name,map):
        self.es_client.indices.create(index=index_name,body=map)
    # TODO:进行删除一个数据库，即index
    def delete_es_index(self, index_name):
        self.es_client.indices.delete(index=index_name)

    # 数据库不用进入，也不用退出。


class LoadElasticSearch(object):
    # TODO:对单个index进行增删改查
    def __init__(self, index, doc_type='docx'):
        # TODO:输入单个index的名称
        self.index = index
        self.doc_type = doc_type
        try:
            self.es_client = ElasticSearchClient().es_client
        except Exception as e:
            print(e)
            print('连接es失败，请查看是否连接。')

        if not self.es_client.indices.exists(index=index):
            # 创建Index
            self.es_client.indices.create(index=self.index)

    def set_index_mapping(self, set_mappings):
        # TODO:设置mapping结构
        """
        设置index的mapping，类似于表结构。
        注意！！！！现在仅仅对mapping中的properties参数，其他的参数还很多
        前提为：已有index，并且已自定义分词器，详情见https://blog.csdn.net/u013905744/article/details/80935846
        输入参数举例说明：
            set_mappings = {"answer": {
                        "type": "string",
                        "index": "not_analyzed"
                    },
                    "answerAuthor": {
                        "type": "string"
                    },
                    "answerDate": {
                        "type": "date",
                        "format": "strict_date_optional_time||epoch_millis"//这里出现了复合类型
                    },
                    ...
                    {...
                    }
                }
        """
        mapping = {
            self.doc_type: {
                "properties": set_mappings
            }
        }
        self.es_client.indices.put_mapping(index=self.index, doc_type=self.doc_type, body=mapping)

    def add_date(self, row_obj):
        """
        单条插入ES
        """
        self.es_client.index(index=self.index, doc_type=self.doc_type, body=row_obj)

    def add_date_bulk(self, row_obj_list):
        """
        批量插入ES,输入文本格式为单条插入的list格式
        """
        load_data = []
        i = 1
        bulk_num = 2000  # 2000条为一批
        for row_obj in row_obj_list:
            action = {
                "_index": self.index,
                "_type": self.doc_type,
                "_source": row_obj
            }
            load_data.append(action)
            i += 1
            # 批量处理
            if len(load_data) == bulk_num:
                print('插入', i / bulk_num, '批数据')
                print(len(load_data))
                success, failed = bulk(self.es_client, load_data, index=self.index, raise_on_error=True)
                del load_data[0:len(load_data)]
                print(success, failed)

        if len(load_data) > 0:
            success, failed = bulk(self.es_client, load_data, index=self.index, raise_on_error=True)
            del load_data[0:len(load_data)]
            print(success, failed)

    def update_by_id(self, row_obj):
        """
        根据给定的_id,更新ES文档
        :return:
        """

        _id = row_obj.get("_id", 1)
        row_obj.pop("_id")
        self.es_client.update(index=self.index, doc_type=self.doc_type, body={"doc": row_obj}, id=_id)

    def delete_by_id(self, _id):
        """
        根据给定的id,删除文档
        :return:
        """
        self.es_client.delete(index=self.index, doc_type=self.doc_type, id=_id)

    def search_by_query(self, body):
        '''
        根据查询的query语句，来搜索查询内容
        '''
        search_result = self.es_client.search(index=self.index, doc_type=self.doc_type, body=body)
        return search_result

三、如何使用

1. 创建索引时指定 Mapping

我们在创建索引时，需要给创建的索引指定 Mapping，我将 Mapping 文件放入了一个 xxx.json 文件中

{
  "settings": {
  #设置副本数
   "number_of_replicas": 1,
     #设置分片
   "number_of_shards": 4,
      #设置分析器 我们采用ik作为tokenizer pinyin作为filter
   "analysis": {
     "analyzer": {
       "my_analyzer":{
       "type":"custom",
       "tokenizer":"ik_max_word",
       "filter":["pinyin_first_letter_and_full_pinyin_filter"]
     }
     },
     "filter": {
       "pinyin_first_letter_and_full_pinyin_filter": {
                    "type" : "pinyin",
                    "keep_first_letter" : "true",
                    "keep_full_pinyin" : "false",
                    "keep_none_chinese" : "true",
                    "keep_original" : "false",
                    "limit_first_letter_length" : 16,
                    "lowercase" : "true",
                    "trim_whitespace" : "true",
                    "keep_none_chinese_in_first_letter" : "true"
                }
     }

   }
 },
 "mappings": {
   "dynamic_templates": [
     {
       "strings":{
           #设定读取到索引中是String类型就设置type为text字段采用我自己设置的分析器，并增加 keyword字段
         "match_mapping_type":"string", 
         "mapping":{
           "type":"text",
           "analyzer":"my_analyzer",
           "fields":{
             "raw":{
               "type":"keyword"
             }

           }
         }
       }
     }
     ]
 }
}

创建代码

mappath="xxxx/xxxx.json"
f=open(mappath,'r',encoding='utf-8')
#读取map
map=json.load(f)
es=ElasticSearchClient()
#创建索引
es.createindex_by_map(indexname,map=map)

2. 查询

es_client = LoadElasticSearch(indexname)
search={"query":xxxx}
res = es_client.search_by_query(one_body)