欢迎来真孝善网,为您提供真孝善正能量书籍故事!

深入解析Elasticsearch批量操作技巧与应用

时间:11-08 现代故事 提交错误

这篇文章给大家聊聊关于深入解析Elasticsearch批量操作技巧与应用,以及对应的知识点,希望对各位有所帮助,不要忘了收藏本站哦。

Solution:

我需要解决两个问题:

查询多个Index的内容,然后将符合条件的数据写入新的Index:

这个问题比较简单。 elasticsearch的helpers模块提供了一个完美的方法来做到这一点:reindex()

elasticsearch.helpers.reindex(client,source_index,target_index,query=None,target_client=None,chunk_size=500,scroll=u"5m",scan_kwargs={},bulk_kwargs={}) 该方法的参数提供了source_index( )可以是List)、target_index、query、scroll_size和滚动存储时间,所以直接运行就可以了。

批量更新现有Index中的所有数据,为每个文档添加一个字段并赋值:

官方文档中的api签名如下:

elasticsearch.helpers.bulk(客户端,操作,stats_only=False,** kwargs)

我一直不明白参数动作是什么类型。我认为它是函数类行的参数。然后看源码,发现其实是一个List,而且是要操作的文档的集合。官方文档显示必须满足这个要求。看起来像search() 返回的结果格式: { "_index": "index-name", "_type": "document", "_id": 42, "_parent": 5, "_ttl": "1d" , "_source": { "title": "你好世界!", "body": "." }}

但他也说:Thebulk()

api 接受索引、创建、删除和更新操作。使用_op_type字段指定操作(_op_type默认为index):

{ "_op_type": "删除", "_index": "索引名称", "_type": "文档", "_id": 42,}{ "_op_type": "更新", "_index": "索引- name", "_type": "document", "_id": 42, "doc": {"question": "生命、宇宙和一切。"}}

我将"_op_type":"update" 添加到我的数据中,然后运行时不断出现错误:

TransportError(400, u"action_request_validation_exception",u"验证失败: 1: 脚本或文档丢失

直到我尝试删除"_op_type"字段,它终于成功了。下面是我的代码:

def queryInES(esinstance):

search_body={"query":{"match_all":{}}}

page=esinstance.search(index="my_index", body=search_body, search_type="scan", doc_type="Tweet",scroll="5s", size=1000)

sid=页面["_scroll_id"]

滚动大小=页面["点击数"]["点击数"]

而(scroll_size0):

pg=es.scroll(scroll_id=sid,scroll="30s")

滚动大小=len(pg["hits"]["hits"])

print "滚动大小:" + str(scroll_size)

sid=pg["_scroll_id"]

数据=pg["点击数"]["点击数"]

……

对于我在范围内(0,scroll_size):

data[i]["_source"]["attributes"]["senti"]={"label":label, "score": 分数, "confidence": 置信度}

用户评论

凉笙墨染

我正在学习 Elasticsearch,Bulk API 看起来很有用!

    有16位网友表示赞同!

不相忘

想了解更多关于使用Bulk API优化索引速度的信息。

    有11位网友表示赞同!

怪咖

不知道 Bulk API 是否可以用来更新多个文档?

    有13位网友表示赞同!

葵雨

有没有关于Bulk API错误处理的教程?

    有16位网友表示赞同!