es批量索引

使用Python操作Elasticsearch数据索引的教程

这篇文章主要介绍了使用Python操作Elasticsearch数据索引的教程,Elasticsearch处理数据索引非常高效,要的朋友可以参考下

Elasticsearch是一个分布式、Restful的搜索及分析服务器，Apache Solr一样，它也是基于Lucence的索引服务器，但我认为Elasticsearch对比Solr的优点在于：

  • 轻量级：安装启动方便，下载文件之后一条命令就可以启动；
  • Schema free：可以向服务器提交任意结构的JSON对象，Solr中使用schema.xml指定了索引结构；
  • 多索引文件支持：使用不同的index参数就能创建另一个索引文件，Solr中需要另行配置；
  • 分布式：Solr Cloud的配置比较复杂。

环境搭建

启动Elasticsearch，访问端口在9200，通过浏览器可以查看到返回的JSON数据，Elasticsearch提交和返回的数据格式都是JSON.

?


1

>> bin/elasticsearch -f

安装官方提供的Python API，在OS X上安装后出现一些Python运行错误，是因为setuptools版本太旧引起的，删除重装后恢复正常。

?


1

>> pip install elasticsearch

索引操作

对于单条索引，可以调用create或index方法。

?


1

2

3

4

5

from datetime import datetime

from elasticsearch import Elasticsearch

es = Elasticsearch() #create a localhost server connection, or Elasticsearch("ip")

es.create(index="test-index", doc_type="test-type", id=1,

  body={"any":"data", "timestamp": datetime.now()})

Elasticsearch批量索引的命令是bulk，目前Python API的文档示例较少，花了不少时间阅读源代码才弄清楚批量索引的提交格式。

?


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

from datetime import datetime

from elasticsearch import Elasticsearch

from elasticsearch import helpers

es = Elasticsearch("10.18.13.3")

j = 0

count = int(df[0].count())

actions = []

while (j < count):

   action = {

        "_index": "tickets-index",

        "_type": "tickets",

        "_id": j + 1,

        "_source": {

              "crawaldate":df[0][j],

              "flight":df[1][j],

              "price":float(df[2][j]),

              "discount":float(df[3][j]),

              "date":df[4][j],

              "takeoff":df[5][j],

              "land":df[6][j],

              "source":df[7][j],

              "timestamp": datetime.now()}

        }

  actions.append(action)

  j += 1

  if (len(actions) == 500000):

    helpers.bulk(es, actions)

    del actions[0:len(actions)]

if (len(actions) > 0):

  helpers.bulk(es, actions)

  del actions[0:len(actions)]

在这里发现Python API序列化JSON时对数据类型支撑比较有限，原始数据使用的NumPy.Int32必须转换为int才能索引。此外，现在的bulk操作默认是每次提交500条数据，我修改为5000甚至50000进行测试，会有索引不成功的情况。

?


1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

#helpers.py source code

def streaming_bulk(client, actions, chunk_size=500, raise_on_error=False,

    expand_action_callback=expand_action, **kwargs):

  actions = map(expand_action_callback, actions)

  # if raise on error is set, we need to collect errors per chunk before raising them

  errors = []

  while True:

    chunk = islice(actions, chunk_size)

    bulk_actions = []

    for action, data in chunk:

      bulk_actions.append(action)

      if data is not None:

        bulk_actions.append(data)

    if not bulk_actions:

      return

def bulk(client, actions, stats_only=False, **kwargs):

  success, failed = 0, 0

  # list of errors to be collected is not stats_only

  errors = []

  for ok, item in streaming_bulk(client, actions, **kwargs):

    # go through request-reponse pairs and detect failures

    if not ok:

      if not stats_only:

        errors.append(item)

      failed += 1

    else:

      success += 1

  return success, failed if stats_only else errors

对于索引的批量删除和更新操作，对应的文档格式如下，更新文档中的doc节点是必须的。

?


1

2

3

4

5

6

7

8

9

10

11

12

13

{

  ‘_op_type‘: ‘delete‘,

  ‘_index‘: ‘index-name‘,

  ‘_type‘: ‘document‘,

  ‘_id‘: 42,

}

{

  ‘_op_type‘: ‘update‘,

  ‘_index‘: ‘index-name‘,

  ‘_type‘: ‘document‘,

  ‘_id‘: 42,

  ‘doc‘: {‘question‘: ‘The life, universe and everything.‘}

}

常见错误

  • SerializationError：JSON数据序列化出错，通常是因为不支持某个节点值的数据类型
  • RequestError：提交数据格式不正确
  • ConflictError：索引ID冲突
  • TransportError：连接无法建立

性能

上面是使用MongoDB和Elasticsearch存储相同数据的对比，虽然服务器和操作方式都不完全相同，但可以看出数据库对批量写入还是比索引服务器更具备优势。

Elasticsearch的索引文件是自动分块，达到千万级数据对写入速度也没有影响。但在达到磁盘空间上限时，Elasticsearch出现了文件合并错误，并且大量丢失数据（共丢了100多万条），停止客户端写入后，服务器也无法自动恢复，必须手动停止。在生产环境中这点比较致命，尤其是使用非Java客户端，似乎无法在客户端获取到服务端的Java异常，这使得程序员必须很小心地处理服务端的返回信息。

时间： 2024-11-09 08:24:23

es批量索引的相关文章

ES重建索引(reindex)性能优化建议

Reindex官方文档 https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-reindex.html Reindex简介 5.X版本后新增Reindex.Reindex可以直接在Elasticsearch集群里面对数据进行重建,如果你的mapping因为修改而需要重建,又或者索引设置修改需要重建的时候,借助Reindex可以很方便的异步进行重建,并且支持跨集群间的数据迁移.比如按天创建的索引可以定期重建合并到以

ES 服务器 索引、类型仓库基类 BaseESStorage

/******************************************************* * * 作者:朱皖苏 * 创建日期:20180508 * 说明:此文件只包含一个类,具体内容见类型注释. * 运行环境:.NET 4.0 * 版本号:1.0.0 * * 历史记录: * 创建文件 朱皖苏 20180508 14:00 * *******************************************************/ using DBEN.Basic.

在ES批量插入数据超时时自动重试

当我们使用ES批量插入数据的时候,一般会这样写代码: from elasticsearch import Elasticsearch,helpers es =Elasticsearch(hosts=[{'host':'localhost','port':9200}]) def gendata(): mywords =['foo','bar','baz'] for word in mywords: yield {"_index":"mywords","_ty

Python操作es批量读取数据

1. Python连接elasticserach python连接elasticsearch有一下几种连接方式 pip3 instal elasticsearch from elasticsearch import Elasticsearch es = Elasticsearch() # 默认连接本地elasticsearch es = Elasticsearch(["127.0.0.1:9200"]) # 连接本地9200端口 es = Elasticsearch(["19

Elasticsearch Bulk API批量索引

这篇博客介绍一下Elasticsearch对多个文档进行索引的简便方法.Bulk api的支持可以实现一次请求执行批量的添加.删除.更新等操作.Bulk操作使用的是UDP协议,UDP无法确保与ElasticSearch服务器通信时不丢失数据. 一.Bulk API 使用bulk命令时,REST API以_bulk结尾,批量操作写在json文件中,官网给出的语法格式: action_and_meta_data\n optional_source\n action_and_meta_data\n o

es备份索引

1.解压https://github.com/medcl/esm-abandonedhttps://github.com/medcl/esm-abandoned/releases tar xf linux64.tar.gz 1.备份escd /root/bin/linux64/./esm -s http://10.33.8.103:9201 -x "petition_data" -b 5 --count=5000 --sliced_scroll_size=10 --refresh -o

Luke 5—— 可视化 Lucene 索引查看工具，可以查看ES的索引

Luke 5 发布,可视化 Lucene 索引查看工具  oschina 发布于2015年08月31日 这是一个主要版本,该版本支持 Lucene 5.2.0. 它支持 elasticsearch 1.6.0(Lucene的4.10.4) 已解决的问题:#20增加支持重建索引并不会存储领域,不暴露位置的字段值. Pull Requests:#23 Elasticsearch 支持和阴影插件组装#26 添加 .gitignore 文件#27 支持 Lucene 5#28 luke.sh 新增LUK

elasticsearch批量索引数据示例

示例数据文件document.json(index表示在索引中增加或替换现有文档,create表示如果文档不存在则添加文档,delete表示删除文档): { "index": { "_index": "addr", "_type": "contact", "_id": 1 }}{ "name": "Fyodor Dostoevsky", "

es学习-索引管理

1.创建索引 http://192.168.0.108:9200/suoyinguanli211/ 参数: { "settings":{ "index":{ "number_of_shards":3,分片数 "number_of_replicas":2 副本数 } } } 结果如图 修改索引信息 url:http://192.168.0.108:9200/suoyinguanli211/_settings/ 参数: { &qu