我想将一堆 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}

最佳答案

同时,我发现了以至少合理的速度执行此操作的多种可能性:

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'

在大型数据集上运行此代码时,可以通过安装pythont的默认json库(用ujsonrapidjson分别替换import ujson as jsonimport rapidjson as json)来节省几分钟。

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

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

10-05 20:31
查看更多