我想将一堆 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
库(用ujson或rapidjson分别替换import ujson as json
或import rapidjson as json
)来节省几分钟。通过将步骤的顺序执行替换为并行步骤,可以实现更大的加速,从而在请求等待Elasticsearch处理所有文档并返回响应时,读取和转换不会停止。这可以通过线程,多处理,Asyncio,任务队列等来完成,但这不在此问题的范围内。
如果您碰巧找到一种更快地执行to-json-conversion的方法,请告诉我。