使用 python 将字典或 JSON 文件列表导入弹性搜索答案

【问题标题】：Import list of dicts or JSON file to elastic search with python使用 python 将字典或 JSON 文件列表导入弹性搜索
【发布时间】：2017-07-26 04:55:51
【问题描述】：

我有一个 .json.gz 文件，我希望将其加载到弹性搜索中。

我的第一次尝试是使用 json 模块将 JSON 转换为字典列表。

import gzip
import json
from pprint import pprint
from elasticsearch import Elasticsearch

nodes_f = gzip.open("nodes.json.gz")
nodes = json.load(nodes_f)

字典示例：

pprint(nodes[0])

{u'index': 1,
 u'point': [508163.122, 195316.627],
 u'tax': u'fehwj39099'}

使用 Elasticsearch：

es = Elasticsearch()

data = es.bulk(index="index",body=nodes)

但是，这会返回：

elasticsearch.exceptions.RequestError: TransportError(400, u'illegal_argument_exception', u'Malformed action/metadata line [1], expected START_OBJECT or END_OBJECT but found [VALUE_STRING]')

除此之外，我希望能够为给定的point 查询找到tax，以防这会影响我应该如何使用elasticsearch 索引数据。

【问题讨论】：

标签： python json elasticsearch

【解决方案1】：

Alfe 为我指明了正确的方向，但我无法让他的代码正常工作。

我找到了两个解决方案：

用for循环逐行：

es = elasticsearch.Elasticsearch()

for node in nodes:
    _id = node['index']
    es.index(index='nodes',doc_type='external',id=_id,body=node)

批量使用helper:

actions = [
    {
    "_index" : "nodes_bulk",
    "_type" : "external",
    "_id" : str(node['index']),
    "_source" : node
    }
for node in nodes
]

helpers.bulk(es,actions)

对于343724 dicts 列表，Bulk 大约快 22 倍。

【讨论】：

【解决方案2】：

ES批量库出现了几个问题，包括性能问题，无法设置具体的_ids等。但是由于ES的批量API不是很复杂，我们自己做了：

import requests

headers = { 'Content-type': 'application/json',
            'Accept': 'text/plain'}

jsons = []
for d in docs:
   _id = d.pop('_id')  # take _id out of dict
   jsons.append('{"index":{"_id":"%s"}}\n%s\n' % (_id, json.dumps(d)))
data = ''.join(jsons)
response = requests.post(url, data=data, headers=headers)

我们需要设置一个特定的_id，但我想你可以跳过这部分，以防你想由 ES 自动设置一个随机的_id。

希望对您有所帮助。

【讨论】：

谢谢@alfie。你能注释这个吗？例如，我现在无法理解docs 是什么。
docs 是文档列表（dicts）；您的 nodes 似乎是相同的结构。 jsons 包含这些文档的 json 编码版本列表，每个版本都由 ES 的批量 API 需要的内容、包含批量任务（“索引这个！”）的小型单行 json 字典和其他信息引入， e. G。应该用什么_id 对该文档进行索引。
好的。你为什么使用requests？正如我所说，有问题的数据是从文档中读取的，无需通过url 请求
requests 模块只是将数据传输到 ES 以供索引的一种方式。它替换了您代码中的 es.bulk() 调用。

【解决方案3】：

这是我使用批量 api 的工作代码：

定义一个字典列表：

from elasticsearch import Elasticsearch, helpers
es = Elasticsearch([{'host':'localhost', 'port': 9200}])

doc = [{'_id': 1,'price': 10, 'productID' : 'XHDK-A-1293-#fJ3'},
   {'_id':2, "price" : 20, "productID" : "KDKE-B-9947-#kL5"}, 
   {'_id':3, "price" : 30, "productID" : "JODL-X-1937-#pV7"},
   {'_id':4, "price" : 30, "productID" : "QQPX-R-3956-#aD8"}]

helpers.bulk(es, doc, index='products',doc_type='_doc', request_timeout=200)

【讨论】：

为什么他们在未来的版本中删除 doc_type ！！，type 只不过是 table