在没有elasticsearch-py的情况下将pandas数据框索引到Elasticsearch

一尘不染

在没有elasticsearch-py的情况下将pandas数据框索引到Elasticsearch

elasticsearch

我想将一堆pandas数据框（大约一百万行和50列）索引到Elasticsearch中。

当寻找有关如何执行此操作的示例时，大多数人会使用elasticsearch-py的bulk
helper方法，并向其传递处理连接的Elasticsearch类的实例以及使用pandas的dataframe创建的字典列表。orient
=’records’）方法。可以预先将元数据作为新列插入数据框，例如，df['_index'] = 'my_index'等等。

但是，我有理由不使用elasticsearch-
py库，而是想直接与Elasticsearch批量API对话，例如通过请求或其他方便的HTTP库。此外，df.to_dict()不幸的是，在大型数据帧上速度非常慢，并将数据帧转换为字典列表，然后通过elasticsearch-
py将其序列化为JSON听起来像不必要的开销，例如dataframe.to_json（）甚至在大数据框。

将熊猫数据框转换为批量API所需格式的简便快捷方法是什么？我认为朝着正确方向迈出的一步dataframe.to_json()如下：

import pandas as pd
df = pd.DataFrame.from_records([{'a': 1, 'b': 2}, {'a': 3, 'b': 4}, {'a': 5, 'b': 6}])
df
   a  b
0  1  2
1  3  4
2  5  6
df.to_json(orient='records', lines=True)
'{"a":1,"b":2}\n{"a":3,"b":4}\n{"a":5,"b":6}'

现在，这是一个以换行符分隔的JSON字符串，但是，它仍然缺少元数据。将其放入其中的一种表演方式是什么？

编辑： 为完整起见，元数据JSON文档将如下所示：

{"index": {"_index": "my_index", "_type": "my_type"}}

因此，最后，批量API期望的整个JSON如下所示（在最后一行之后附加换行符）：

{"index": {"_index": "my_index", "_type": "my_type"}}
{"a":1,"b":2}
{"index": {"_index": "my_index", "_type": "my_type"}}
{"a":3,"b":4}
{"index": {"_index": "my_index", "_type": "my_type"}}
{"a":5,"b":6}

阅读 377

2020-06-22

共1个答案

一尘不染

同时，我发现了多种可能性，至少以合理的速度如何做到这一点：

import json
import pandas as pd
import requests

# df is a dataframe or dataframe chunk coming from your reading logic
df['_id'] = df['column_1'] + '_' + df['column_2'] # or whatever makes your _id
df_as_json = df.to_json(orient='records', lines=True)

final_json_string = ''
for json_document in df_as_json.split('\n'):
    jdict = json.loads(json_document)
    metadata = json.dumps({'index': {'_id': jdict['_id']}})
    jdict.pop('_id')
    final_json_string += metadata + '\n' + json.dumps(jdict) + '\n'

headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
r = requests.post('http://elasticsearch.host:9200/my_index/my_type/_bulk', data=final_json_string, headers=headers, timeout=60)

除了使用熊猫的to_json()方法，还可以使用to_dict()以下方法。这在我的测试中稍慢一些，但并没有很多：

dicts = df.to_dict(orient='records')
final_json_string = ''
for document in dicts:
    metadata = {"index": {"_id": document["_id"]}}
    document.pop('_id')
    final_json_string += json.dumps(metadata) + '\n' + json.dumps(document) + '\n'

当大数据集运行此，人们可以通过更换Python的默认保存了两三分钟json与库ujson或rapidjson通过安装它，然后import ujson as json或import rapidjson as json分别。

通过将步骤的顺序执行替换为并行步骤，可以实现更大的加速，从而在请求等待Elasticsearch处理所有文档并返回响应时，读取和转换不会停止。这可以通过线程，多处理，Asyncio，任务队列等来完成，但这不在此问题的范围内。

如果您碰巧找到一种更快地执行to-json-conversion的方法，请告诉我。

2020-06-22