Elasticsearch基础及python操作

布满荆棘的人生 2024-04-08 08:58 89阅读 0赞

## 一、ES基础 ##

官方文档：[Elasticsearch: 权威指南 | Elastic][Elasticsearch_ _ _ Elastic]

### 1、名词对比mysql ###

ElasticSearch

MySQL

Index

索引

Database

数据库

Type

类型

Table

表

Document

文档

Row

行

Field

字段

Column

列

Mapping

处理规则

Schema

关系

> **索引库（indices)**：indices是index的复数，代表许多的索引，等价于关系型数据库中的数据库（DataBase）
> 
> **类型（type）**：类型是模拟mysql中数据库下面有N张表（Table）的概念，一个索引库下可以有不同类型的索引，比如商品索引，订单索引，其数据格式不同。不过这会导致索引库混乱，因此未来版本中会移除这个概念
> 
> **文档（document）**：存入索引库原始的数据。比如每一条商品信息，就是一个文档，等价于一个数据库表（Table）下的数据由多行（ROW）
> 
> **字段（field）**：文档中的属性，相当于一个数据库表（Table）下的多列（column，属性）
> 
> **映射配置（mappings）**：字段的数据类型、属性、是否索引、是否存储等特性等价于在一个关系型数据库里面，[schema][]定义了表、每个表的字段，还有表和字段之间的关系

待完善

*  分片（shard）：数据拆分后的各个部分  
    副本（replica）：每个分片的复制

详细参考：[elasticsearch架构及名词解释\_时间总在嘲笑i的博客-CSDN博客][elasticsearch_i_-CSDN]

### 2、操作 对比mysql ###

>  *  在数据库中的增insert、删delete、改update、查select操作等价于ES中的增PUT、删Delete、改POST、查GET
>  *  MySQL 中的 group by、avg、sum 等函数类似于 ES 中的 Aggregations 的部分特性。
>  *  MySQL 中的去重 distinct 类似 ES 中的 cardinality 操作。

### 3、常用语句 ###

#### 新建索引 ####

PUT /my_index
    {
        "settings": { ... any settings ... },
        "mappings": {
            "type_one": { ... any mappings ... },
            "type_two": { ... any mappings ... },
            ...
        }
    }
    
    例：
    PUT es_index
    {
      "settings": {
        "index": {
          "number_of_shards": "10",  //分10个片
          "number_of_replicas": "1"  //1个备份
        }
      },
      "mappings": {    
        "properties":{
            "message":{
                "type":"string"
            },
            "price":{
                "type":"string"
            },
            "tid":{
                "type":"string"
            },
            "user":{
                "type":"string"
            }
        }
    }

[详细参考：elasticsearch mapping设置\_时间总在嘲笑i的博客-CSDN博客\_elasticsearch mapping 时区][elasticsearch mapping_i_-CSDN_elasticsearch mapping]

#### 查看映射关系 ####

GET /索引名/_mapping

#### 查询索引的设置 ####

GET /索引名/_settings

#### 新建/更新数据 ####

PUT /{index}/{type}/{id}
    {
      "field": "value",
      ...
    }
    
    例：
    PUT /weather/d1/1
    {
    "age":1,
    "name":"zs",
    "bri":"2018-08-08"
    }

#### 删除一条指定数据 ####

DELETE  /{index}/{type}/{id}
    {
      "field": "value",
      ...
    }
    
    例：
    DELETE  weather/d1/2

#### 删除索引 ####

#删除一个索引
    DELETE /my_index
    #删除多个索引
    DELETE /index_one,index_two
    DELETE /index_*
    #删除所有索引
    DELETE /_all
    DELETE /*

#### 查询数据 ####

#查询一条数据
    GET {index}/{type}/{id}
    #查询所有数
    GET {index}/_search
    #条件查询
    GET {index}/_search?q=Field:value

对于查询返回结果的解释：[空搜索 | Elasticsearch: 权威指南 | Elastic][_ Elasticsearch_ _ _ Elastic]

#### json-查询 ####

GET {index}/_search
    {
        "query":{
          "match_all": {}
        }
    }

更多参考：[深入搜索 | Elasticsearch: 权威指南 | Elastic][_ Elasticsearch_ _ _ Elastic 1]

## 二、 python操作es ##

官方文档：[Python Elasticsearch Client — Elasticsearch 7.16.3 documentation][Python Elasticsearch Client _ Elasticsearch 7.16.3 documentation]

### 1、安装elasticsearch ###

pip installelasticsearch

### 2、连接 ###

from elasticsearch import Elasticsearch
     
     
    class ElasticSearchClass(object):
     
        def __init__(self, host, port, user, passwrod):
          
            http_auth = user + ":" + passwrod 
            self.es = Elasticsearch(
                [host],
                port=port,
                http_auth=http_auth,
                # sniff_on_start=True,  # 启动前嗅探es集群服务器
                sniff_on_connection_fail=True,  # es集群服务器节点连接异常时是否刷新es节点信息
                sniff_timeout=60  # 每60秒刷新节点信息
            )

### 3、常用操作 ###

#### 创建索引 ####

@retry(tries=3)
        def create_index(self, index_name, mapping):
    
            self.es.indices.create(index=index_name, ignore=400)
            self.es.indices.put_mapping(index=index_name, body=mapping)

#### 插入一条数据 ####

def insert(self, index, type, body, id=None):
            '''
            插入一条body给指定的index、指定的type下;
            可指定Id,若不指定,ES会自动生成
            :param index: 待插入的index值      
            :param body: 待插入的数据  # dict型
            :param id: 自定义Id值
            :return:
            '''
            return self.es.index(index=index,doc_type=type,body=body,id=id,request_timeout=30)

#### 查询一条数据 ####

def get(self, doc_type, indexname, id):
            # index中具体的一条
            return self.es.get(index=indexname,doc_type=doc_type, id=id)

#### 根据索引进行搜索 ####

#通过index搜索数据  其中，搜索之后数据显示默认为十条数据
    def searchindex(self, index):
            """
            查找所有index数据
            """
            try:
                return self.es.search(index=index)
            except Exception as err:
                print(err)

#### 查找index下所有符合条件的数据 ####

def searchDoc(self, index=None, type=None, body=None):
            '''
            查找index下所有符合条件的数据
            :param index:
            :param type:
            :param body: 筛选语句,符合DSL语法格式
            :return:
            '''
            return self.es.search(index=index, doc_type=type, body=body)

#### 更新其中具体的一条 ####

def update(self, doc_type, indexname,  body=data, id):
            # 更新其中具体的一条
            return self.es.update(index=indexname,doc_type=doc_type,  body=data, id=id)

#### 统计类操作 ####

def count(self, indexname):
            """
            :return: 统计index总数
            """
            return self.conn.count(index=indexname)
    
    def proportion_not_null(self, index, field=None):
            """非空统计"""
            a = self.count(index)['count']
            b = self.count(index, {'query': {'bool': {'must': {'exists': {'field': 
                                                                         field}}}}})['count']
            print(field, a, b, b / a)
     
        def aggs_terms(self, index, field, size=15):
            """单字段统计"""
            return self.search({
                'aggs': {
                    'CUSTOM NAME': {
                        'terms': {
                            'field': field,
                            'size': size,  # 解决aggs显示不全
                        }
                    }
                }
            }, index)['aggregations']['CUSTOM NAME']['buckets']

#### 删除索引下所有的数据 ####

def clear_index(index):
        body ={
            "query": {
                "match_all": {
                }
            }
        }
    
        self.es.delete_by_query(index,body)

#### 删除索引中具体某一条 ####

#数据删除时候是根据id进行删除 
    def delete(self, indexname, doc_type, id):
            """
            :param indexname:
            :param doc_type:
            :param id:
            :return: 删除index中具体的一条
            """
            self.es.delete(index=indexname, doc_type=doc_type, id=id)

#### 删除所有索引 ####

def delete_index(self, index):
            # 删除所有索引
            if not index：
                for index in es.indices.get("*"):
                    es.indices.delete(index)
            else：
                es.indices.delete(index)

## 三、安装requests模块，通过GET、POST方式操作Elasticsearch ##

参考：[python Elasticsearch5.x使用 - shhnwangjian - 博客园][python Elasticsearch5.x_ - shhnwangjian -]

[Elasticsearch_ _ _ Elastic]: https://www.elastic.co/guide/cn/elasticsearch/guide/current/index.html
[schema]: https://so.csdn.net/so/searchq=schema
[elasticsearch_i_-CSDN]: https://blog.csdn.net/weixin_42055771/article/details/104040373
[elasticsearch mapping_i_-CSDN_elasticsearch mapping]: https://blog.csdn.net/weixin_42055771/article/details/104042464
[_ Elasticsearch_ _ _ Elastic]: https://www.elastic.co/guide/cn/elasticsearch/guide/current/empty-search.html#_hits
[_ Elasticsearch_ _ _ Elastic 1]: https://www.elastic.co/guide/cn/elasticsearch/guide/current/search-in-depth.html
[Python Elasticsearch Client _ Elasticsearch 7.16.3 documentation]: https://elasticsearch-py.readthedocs.io/en/v7.16.3/
[python Elasticsearch5.x_ - shhnwangjian -]: https://www.cnblogs.com/shhnwangjian/p/7160902.html