ElasticSearch

索引(index)

索引(index)就是一个有相似特征的文档的集合，es把数据存储到一个或多个索引中。通常index是一个比较大的分类，比如按照不同的客户，或者不同的产品，或者不同的时间戳。

通常构造index的时候需要考虑好分类的粒度，因为如果后面想要批量删除的话基本单位是索引。如果构造的index分类粒度太大，那么批量删除的时候将会变得很麻烦。

如果是采用ES来搭建实时日志搜索平台，那么通常建议index按照时间戳来进行命名，因为这样的话可以设置超过几天的日志全部删除，清理磁盘空间。

文档(document)

文档(document)是es中存储数据的实体，它是可以被索引的基本单元。每一个文档都以json格式表示，每一个索引可以任意多个文档。

简单来说就是文档其实就是真实的每一条的存储的数据，比如每条以json格式表示的日志。每一个文档物理上存在一个索引中，但文档必须有一个文档类型(type)。

类型(`type`)

类型(type)是索引之下的逻辑分区，相对index更细的分类粒度。比如如果index以日期命名，那么type可以以具体的客户命名。然后每一个客户下面，又有很多的文档实体。

`id`

每一个文档实体，ES会自动为其创建一个唯一的id，比如“AV8tQS119lFHJS7Mv7v8”。

这里不太重要不需要关注，但有一点需要注意，日志的字段不能有id这个字段，否则会跟自动创建的id冲突，导致可能插入数据失败。

单一操作

插入　　create：必须指定待查询的idnex、type、id和查询体body；缺一不可，否则报错　　index：相比于create，index的用法就相对灵活很多；id并非是一个必选项，如果指定，则该文档的id就是指定值，若不指定，则系统会自动生成一个全局唯一的id赋给该文档。 eg：

body = {"name": 'lucy', 'sex': 'female', 'age': 10}
es = Elasticsearch(['localhost:9200'])
es.index(index='indexName', doc_type='typeName', body, id=None)

删除　　delete：删除指定index、type、id的文档

es.delete(index='indexName', doc_type='typeName', id='idValue')

查找　　get：获取指定index、type、id所对应的文档

es.get(index='indexName', doc_type='typeName', id='idValue')

更新　　update：跟新指定index、type、id所对应的文档

es.update(index='indexName', doc_type='typeName', id='idValue', body={待更新字段})

批量操作

条件查询　　search：查询满足条件的所有文档，没有id属性，且index，type和body均可为None。　　body的语法格式必须符合DSL（Domain Specific Language ）格式

query = {'query': {'match_all': {}}}# 查找所有文档

query = {'query': {'term': {'name': 'jack'}}}# 查找名字叫做jack的所有文档

query = {'query': {'range': {'age': {'gt': 11}}}}# 查找年龄大于11的所有文档

allDoc = es.search(index='indexName', doc_type='typeName', body=query)

print allDoc['hits']['hits'][]# 返回第一个文档的内容

条件删除　　delete_by_query：删除满足条件的所有数据，查询条件必须符合DLS格式

query = {'query': {'match': {'sex': 'famale'}}}# 删除性别为女性的所有文档

query = {'query': {'range': {'age': {'lt': 11}}}}# 删除年龄小于11的所有文档

es.delete_by_query(index='indexName', body=query, doc_type='typeName')

条件更新　　update_by_query：更新满足条件的所有数据，写法同上删除和查询
批量插入、删除、更新　　bulk：在这重点和大家聊聊bulk方法，前面的所有方法都很简单，唯独这个bulk在笔者开始接触的时候，花费了不少时间；这个方法可以同时执行多个操作，单只请求一次，从而在批量操作的时候，可以很大程度上减少程序系统开销。此外，bulk不仅可以一次性批量执行插入、或者删除操作，还可以在一次请求中，既可以插入、又可以删除和更新操作。　　但是需要注意的是，任何一种操作都有固定的文档格式，只有完全符合该格式要求，才可执行成功。废话不多说，直接上代码：

doc = [
     {"index": {}},
     {'name': 'jackaaa', 'age': 2000, 'sex': 'female', 'address': u'北京'},
     {"index": {}},
     {'name': 'jackbbb', 'age': 3000, 'sex': 'male', 'address': u'上海'},
     {"index": {}},
     {'name': 'jackccc', 'age': 4000, 'sex': 'female', 'address': u'广州'},
     {"index": {}},
     {'name': 'jackddd', 'age': 1000, 'sex': 'male', 'address': u'深圳'},
 ]
 doc = [
    {'index': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
    {'name': 'jack', 'sex': 'male', 'age': 10 }
    {'delete': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
    {"create": {'_index' : 'indexName', "_type" : 'typeName', '_id': 'idValue'}}
    {'name': 'lucy', 'sex': 'female', 'age': 20 }
    {'update': {'_index': 'indexName', '_type': 'typeName', '_id': 'idValue'}}
    {'doc': {'age': '100'}}
 ]
 es.bulk(index='indexName'，  doc_type='typeName', body=doc)

创建索引

# -.- coding:utf-8 -.-
from __future__ import print_function
from pprint import pprint
from elasticsearch import Elasticsearch

es_hosts = ["192.168.9.119:9200"]
index_name = "log"
doc_type = "20170103"


def main():
    es = Elasticsearch(es_hosts)
    es.indices.create(index=index_name, body={"mappings":{doc_type: {"properties": {"name": {"type": "text"}, "gender":{"type": "text"}, "age":{"type": "integer"}, "phone":{"type": "keyword"}}}}})
    res = es.search(index=index_name, body={"query": {"match_all": {}}})
    pprint(res)
    # pprint(es.info())


if __name__ == '__main__':
    main()

查询创建的索引

{
    "log": {
        "aliases": {},
        "mappings": {
            "20170103": {
                "properties": {
                    "age": {
                        "type": "integer"
                    },
                    "gender": {
                        "type": "text"
                    },
                    "name": {
                        "type": "text"
                    },
                    "phone": {
                        "type": "keyword"
                    }
                }
            }
        },
        "settings": {
            "index": {
                "creation_date": "1512980895137",
                "number_of_shards": "5",
                "number_of_replicas": "1",
                "uuid": "TOrOEfoHQiSKX8oqlZ6URw",
                "version": {
                    "created": "5050099"
                },
                "provided_name": "log"
            }
        }
    }
}

你也可以先创建索引，然后创建type再设置mapping

def main():
    es = Elasticsearch(es_hosts)
    # es.indices.create(index="students")
    es.indices.put_mapping(index="students", doc_type="yinianji", body={"properties": {"name": {"type": "text"}}})
    res = es.search(index=index_name, body={"query": {"match_all": {}}})
    pprint(res)
    # pprint(es.info())


if __name__ == '__main__':
    main()

2. 插入数据

index (单条插入)

from datetime import datetime
print(datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S'))
data = {
        "filename": '1233445',
        "url": '/root',
        "status": ,
        "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        #"date": datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S'),
        "total": 100,
        "county": '111111 111112',
        "gender": ,
        "agelow": 12,
        "agehigh": 18
}
es.index(index = index_name, doc_type = doc_type, id= '1233445', body = data)

bulk (批量插入)

# -.- coding:utf-8 -.-
from __future__ import print_function
from pprint import pprint
from elasticsearch import Elasticsearch
from elasticsearch import helpers

es_hosts = ["192.168.9.119:9200"]
index_name = "log"
doc_type = "20170103"
body = []
for i in range(10):
    body.append({
        "_index": "students",
        "_type": "yinianji",
        "_id": i + 1,
        "_source": {
              "name": 'weishihao'
        }
        })

def main():
    es = Elasticsearch(es_hosts)
    helpers.bulk(es, body)
    res = es.search(index='students', body={"query": {"match_all": {}}})
    pprint(res)
    # pprint(es.info())


if __name__ == '__main__':
    main()

3. 修改mapping结构

在elasticsearch中，更改mapping结构只能新增field。所以

es.indices.put_mapping(
                index=index_name,
                doc_type=doc_type,
                body={
                    "properties": {
                         "county": {"type": "text"},
                         "total" : {"type": "integer"},
                         "gender": {"type": "integer"},
                         "agelow": {"type": "integer"},
                         "agehigh": {"type": "integer"}
                    }
                }
            )

4. 查询数据

# -.- coding:utf-8 -.-
from __future__ import print_function
from pprint import pprint
from elasticsearch import Elasticsearch
from elasticsearch import helpers

es_hosts = ["192.168.9.119:9200"]
index_name = "log"
doc_type = "20170103"
body = []
for i in range(10):
    body.append({
        "_index": "students",
        "_type": "yinianji",
        "_id": i + 1,
        "_source": {
              "name": 'weishihao'
        }
        })

def main():
    es = Elasticsearch(es_hosts)
    # helpers.bulk(es, body)
    res = es.search(index='students', doc_type='yinianji', body={"query": {"match_all": {}}})
    pprint(res)
    # pprint(es.info())


if __name__ == '__main__':
    main()

返回值中的total值会给出总数据量，但是_source中，返回显示的，默认只有10条那么，我们如何查询所有的数据呢，

es = Elasticsearch(es_hosts)
 scanResp = helpers.scan(es, {"query": {"match_all": {}}}, index= 'quanguorenkou', scroll= "10m") 
    for hit in scanResp:  
        print(hit)

这么我们就可以查询所有的数据了。但是如果数据量比较大的时候，我们仅仅需要部分的数据的话，可以指定查询条件，比如：前缀查询：

scanResp = helpers.scan(es, {"query": {"prefix": {"studentid": {"value": "330"}}}}, index= 'students', scroll= "100m", size=40000)

上述语句实现的功能是查询studentid字段以330开头的所有文档。

5. 修改数据

es.update(index = index_name, doc_type = doc_type, id='1233445', body={"script": "ctx._source.status = 1"})
复制代码
6.删除索引
es.indices.delete(index=index)

7.一个完整的例子

# -.- coding:utf-8 -.-
from __future__ import print_function
from elasticsearch import Elasticsearch, helpers
from pprint import pprint
import sys
import os
sys.path.append(os.path.abspath(os.path.pardir))


from multiprocessing import current_process, Pool
from collections import deque
import time
import re
es_hosts = ["192.168.31.13"]
es_auth = ("elastic", "changeme")
index_name = 'exportdata'
doc_type = 'output'
es = Elasticsearch(es_hosts, http_auth = es_auth)
es.indices.delete(index=index_name)
es.indices.create(index=index_name, body=
                      {"mappings":{doc_type:
                              {"properties": {
                                     "filename": {"type": "text"},
                                     "url":{"type": "text"},
                                     "status":{"type": "integer"},
                                     "date": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"},
                                     "county": {"type": "text"},
                                     "total" : {"type": "integer"},
                                     "gender": {"type": "integer"},
                                     "agelow": {"type": "integer"},
                                     "agehigh": {"type": "integer"}
                                    }
                              }
                     }})
#es.indices.put_mapping(
#                index=index_name,
#                doc_type=doc_type,
#                body={
#                    "properties": {
#                         "county": {"type": "text"},
#                         "total" : {"type": "integer"},
#                         "gender": {"type": "integer"},
#                         "agelow": {"type": "integer"},
#                         "agehigh": {"type": "integer"}
#                    }
#               }
#            )
#from datetime import datetime
#print(datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S'))
#data = {
#        "filename": '1233445',
#        "url": '/root',
#        "status": 0,
#        "date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
#        #"date": datetime.strftime(datetime.now(), '%Y-%m-%d %H:%M:%S'),
#        "total": 100,
#        "county": '111111 111112',
#        "gender": 0,
#        "agelow": 12,
#        "agehigh": 18
#}
#es.index(index = index_name, doc_type = doc_type, id= '1233445', body = data)
#es.update(index = index_name, doc_type = doc_type, id='1233445', body={"script": "ctx._s

ITPub博客

ELK-基础系列(八)-Python调取ES API