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

问题描述

我想在Elasticsearch中索引一堆大熊猫数据框(大约一百万行和50列).

I would like to index a bunch of large pandas dataframes (some million rows and 50 columns) into Elasticsearch.

寻找有关如何执行此操作的示例时，大多数人会使用 elasticsearch-py的批量助手方法，将其传递给实例处理连接的Elasticsearch类以及创建的字典列表使用熊猫的dataframe.to_dict(orient ='records')方法.可以预先将元数据作为新列插入数据框，例如df['_index'] = 'my_index'等.

When looking for examples on how to do this, most people will use elasticsearch-py's bulk helper method, passing it an instance of the Elasticsearch class which handles the connection as well as a list of dictionaries which is created with pandas' dataframe.to_dict(orient='records') method. Metadata can be inserted into the dataframe beforehand as new columns, e.g. df['_index'] = 'my_index' etc.

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

However, I have reasons not to use the elasticsearch-py library and would like to talk to the Elasticsearch bulk API directly, e.g. via requests or another convenient HTTP library. Besides, df.to_dict() is very slow on large dataframes, unfortunately, and converting a dataframe to a list of dicts which is then serialized to JSON by elasticsearch-py sounds like unnecessary overhead when there is something like dataframe.to_json() which is pretty fast even on large dataframes.

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

What would be an easy and quick approach of getting a pandas dataframe into the format required by the bulk API? I think a step in the right direction is using dataframe.to_json() as follows:

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字符串，但是，它仍然缺少元数据.将其放入其中的一种表演方式是什么?

This is now a newline-separated JSON string, however, it is still lacking the metadata. What would be a performing way to get it in there?

修改:为了完整起见，元数据JSON文档如下所示:

edit:For completeness, a metadata JSON document would look like that:

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

因此，最后，批量API期望的整个JSON看起来像这个(在最后一行之后有一个换行符):

Hence, in the end the whole JSON expected by the bulk API would look likethis (with an additional linebreak after the last line):

{"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}

推荐答案

同时，我发现了多种可能性，如何以至少合理的速度实现该目标:

Meanwhile I found out multiple possibilities how to do that with at least reasonable speed:

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)

除了使用pandas的to_json()方法外，还可以如下使用to_dict().在我的测试中，这稍微慢一点，但幅度不大:

Instead of using pandas' to_json() method, one could also use to_dict() as follows. This was slightly slower in my tests but not much:

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或.

When running this on large datasets, one can save a couple of minutes by replacing Python's default json library with ujson or rapidjson via installing it, then import ujson as json or import rapidjson as json, respectively.

通过将步骤的顺序执行替换为并行步骤，可以实现更大的加速，从而在请求等待Elasticsearch处理所有文档并返回响应时，读取和转换不会停止.可以通过Threading，Multiprocessing，Asyncio，Task Queues ...来完成，但这超出了这个问题的范围.

An even bigger speedup can be achieved by replacing the sequential execution of the steps with a parallel one so that reading and converting does not stop while requests is waiting for Elasticsearch to process all documents and return a response. This could by done via Threading, Multiprocessing, Asyncio, Task Queues, ... but this is out of the scope of this question.

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

If you happen to find an approach to do the to-json-conversion even faster, let me know.

这篇关于在没有elasticsearch-py的情况下将 pandas 数据框索引到Elasticsearch中的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！