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

Question

我想将一堆大型 pandas 数据帧(大约百万行和 50 列)索引到 Elasticsearch 中。

在查找有关如何执行此操作的示例时，大多数人会使用 elasticsearch-py's bulk helper method , 传递给它一个实例 of the Elasticsearch class它处理连接以及创建的字典列表with pandas' dataframe.to_dict(orient='records') method .元数据可以作为新列预先插入到数据框中，例如df['_index'] = 'my_index' 等

但是，我有理由不使用 elasticsearch-py 库，想和 Elasticsearch bulk API 谈谈直接，例如通过requests或者其他方便的 HTTP 库。此外，df.to_dict() 在大型数据帧上非常慢，不幸的是，将数据帧转换为字典列表，然后通过 elasticsearch-py 序列化为 JSON 听起来像是不必要的开销喜欢dataframe.to_json()即使在大型数据帧上也非常快。

将 pandas 数据框转换为批量 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}

Answer 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) 

除了使用 pandas 的 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、任务队列等来完成，但这超出了这个问题的范围。

如果您碰巧找到一种更快地进行到 json 转换的方法，请告诉我。