在没有 elasticsearch-py 的情况下将 pandas 数据帧索引到 Elasticsearch答案

【问题标题】：Index a pandas dataframe into Elasticsearch without elasticsearch-py在没有 elasticsearch-py 的情况下将 pandas 数据帧索引到 Elasticsearch
【发布时间】：2017-07-03 17:44:27
【问题描述】：

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

【问题讨论】：

您可以为您的示例 DF 发布预期的元数据吗？
好的，请看我的编辑。
我不理解那种格式（结构）——它不是有效的 JSON。你能做一个小测试，尝试使用它的批量 API 将这个小的“JSON”加载到 ElasticSearch 中吗？
是的，这确实不是有效的 JSON，而是多个有效 JSON 文档的换行符列表。不幸的是，this is what elasticsearch's bulk API expects。其背后的原因是批量数据在换行符处被拆分为文档，然后文档可能在文档实际被解析之前被转发到接收节点之外的其他节点。

标签： python pandas elasticsearch

【解决方案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 处理所有文档并返回响应时，读取和转换不会停止。这可以通过 Threading、Multiprocessing、Asyncio、Task Queues 来完成……但这超出了这个问题的范围。

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

【讨论】：

只看这段代码，您正在序列化为 json，然后再次反序列化它以进行循环。我想你可以通过使用 df.iterrows 然后只在行本身上调用 to_json 来获得简单的加速

【解决方案2】：

此函数将 pandas 数据帧插入弹性搜索（逐块）

def insertDataframeIntoElastic(dataFrame,index='index', typ = 'test', server = 'http://localhost:9200',
                           chunk_size = 2000):
    headers = {'content-type': 'application/x-ndjson', 'Accept-Charset': 'UTF-8'}
    records = dataFrame.to_dict(orient='records')
    actions = ["""{ "index" : { "_index" : "%s", "_type" : "%s"} }\n""" % (index, typ) +json.dumps(records[j])
                    for j in range(len(records))]
    i=0
    while i<len(actions):
        serverAPI = server + '/_bulk' 
        data='\n'.join(actions[i:min([i+chunk_size,len(actions)])])
        data = data + '\n'
        r = requests.post(serverAPI, data = data, headers=headers)
        print r.content
        i = i+chunk_size

【讨论】：