将 pandas 数据框索引到 Elasticsearch 中，无需使用 elasticsearch-py

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

在寻找如何执行此操作的示例时，大多数人会使用elasticsearch-py 的批量辅助方法 https://elasticsearch-py.readthedocs.io/en/master/helpers.html#elasticsearch.helpers.bulk，传递一个实例Elasticsearch 类的 https://elasticsearch-py.readthedocs.io/en/master/api.html#elasticsearch它处理连接以及创建的字典列表使用 pandas 的 dataframe.to_dict(orient='records') 方法 http://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.to_dict.html。元数据可以作为新列预先插入到数据框中，例如df['_index'] = 'my_index' etc.

但是，我有理由不使用 elasticsearch-py 库，并想与Elasticsearch 批量 API https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html直接，例如通过requests http://docs.python-requests.org/en/master/或另一个方便的 HTTP 库。除了，df.to_dict()不幸的是，在大型数据帧上速度非常慢，并且将数据帧转换为字典列表，然后由 elasticsearch-py 序列化为 JSON，当存在类似情况时，听起来像是不必要的开销dataframe.to_json() http://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.to_json.html即使在大型数据帧上，这也相当快。

将 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 字符串，但是，它仍然缺少元数据。将其放入其中的执行方式是什么？

edit:为了完整起见，元数据 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) 

而不是使用熊猫'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 https://pypi.python.org/pypi/ujson or 快速JSON https://github.com/kenrobbins/python-rapidjson通过安装它，然后import ujson as json or import rapidjson as json， 分别。

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

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