从 ElasticSearch 结果创建 DataFrame答案

【问题标题】：Creating DataFrame from ElasticSearch Results从 ElasticSearch 结果创建 DataFrame
【发布时间】：2014-09-30 21:49:23
【问题描述】：

我正在尝试在 pandas 中构建一个 DataFrame，使用对 Elasticsearch 进行非常基本的查询的结果。我得到了我需要的数据，但它需要对结果进行切片以构建正确的数据框。我真的只关心获取每个结果的时间戳和路径。我尝试了几种不同的 es.search 模式。

代码：

from datetime import datetime
from elasticsearch import Elasticsearch
from pandas import DataFrame, Series
import pandas as pd
import matplotlib.pyplot as plt
es = Elasticsearch(host="192.168.121.252")
res = es.search(index="_all", doc_type='logs', body={"query": {"match_all": {}}}, size=2, fields=('path','@timestamp'))

这给出了 4 个数据块。 [u'hits', u'_shards', u'took', u'timed_out']。我的结果在点击中。

res['hits']['hits']
Out[47]: 
[{u'_id': u'a1XHMhdHQB2uV7oq6dUldg',
  u'_index': u'logstash-2014.08.07',
  u'_score': 1.0,
  u'_type': u'logs',
  u'fields': {u'@timestamp': u'2014-08-07T12:36:00.086Z',
   u'path': u'app2.log'}},
 {u'_id': u'TcBvro_1QMqF4ORC-XlAPQ',
  u'_index': u'logstash-2014.08.07',
  u'_score': 1.0,
  u'_type': u'logs',
  u'fields': {u'@timestamp': u'2014-08-07T12:36:00.200Z',
   u'path': u'app1.log'}}]

我唯一关心的是获取时间戳和每次点击的路径。

res['hits']['hits'][0]['fields']
Out[48]: 
{u'@timestamp': u'2014-08-07T12:36:00.086Z',
 u'path': u'app1.log'}

我终其一生都无法弄清楚是谁将结果放入 pandas 的数据框中。所以对于我返回的 2 个结果，我希望有一个类似的数据框。

   timestamp                   path
0  2014-08-07T12:36:00.086Z    app1.log
1  2014-08-07T12:36:00.200Z    app2.log

【问题讨论】：

标签： python pandas elasticsearch

【解决方案1】：

或者你可以使用 pandas 的 json_normalize 函数：

from pandas.io.json import json_normalize
df = json_normalize(res['hits']['hits'])

然后按列名过滤结果数据框

【讨论】：

为了提高性能，预先过滤，例如[x['_source'] for x in res['hits']['hits']]
按原样尝试，得到一个错误“'AttrDict' 对象没有属性'values'”。明天早上可能会调查这个

【解决方案2】：

更好的是，您可以使用出色的 pandasticsearch 库：

from elasticsearch import Elasticsearch
es = Elasticsearch('http://localhost:9200')
result_dict = es.search(index="recruit", body={"query": {"match_all": {}}})

from pandasticsearch import Select
pandas_df = Select.from_dict(result_dict).to_pandas()

【讨论】：

我可以知道如何展平df中的嵌套信息吗？

【解决方案3】：

有一个不错的玩具叫pd.DataFrame.from_dict，你可以在这样的情况下使用它：

In [34]:

Data = [{u'_id': u'a1XHMhdHQB2uV7oq6dUldg',
      u'_index': u'logstash-2014.08.07',
      u'_score': 1.0,
      u'_type': u'logs',
      u'fields': {u'@timestamp': u'2014-08-07T12:36:00.086Z',
       u'path': u'app2.log'}},
     {u'_id': u'TcBvro_1QMqF4ORC-XlAPQ',
      u'_index': u'logstash-2014.08.07',
      u'_score': 1.0,
      u'_type': u'logs',
      u'fields': {u'@timestamp': u'2014-08-07T12:36:00.200Z',
       u'path': u'app1.log'}}]
In [35]:

df = pd.concat(map(pd.DataFrame.from_dict, Data), axis=1)['fields'].T
In [36]:

print df.reset_index(drop=True)
                 @timestamp      path
0  2014-08-07T12:36:00.086Z  app2.log
1  2014-08-07T12:36:00.200Z  app1.log

分四步展示：

1、将列表中的每一项（即dictionary）读入DataFrame

2，我们可以将列表中的所有项目通过concat逐行放入一个大的DataFrame，因为我们将为每个项目执行步骤#1，我们可以使用map来完成。

3，然后我们访问标记为'fields'的列

4，如果我们希望索引为默认的int 序列，我们可能希望将DataFrame 旋转90 度（转置）和reset_index。

【讨论】：

非常感谢。这行得通。你能向我解释一下这部分吗？ “pd.concat(map(pd.DataFrame.from_dict, Data), axis=1['fields'].T” 我已经走上了逐一遍历结果的路线，并为每个时间戳/路径创建一个元组，将其附加到列表中，然后使用 from_record 读取该元组列表。您的方式要快得多。
我相信 pandas from_dict 现在可以将 list 作为参数了

【解决方案4】：

我测试了所有答案的性能，发现pandasticsearch 方法在很大程度上是最快的：

测试：

test1（使用 from_dict）

%timeit -r 2 -n 5 teste1(resp)

每个循环 10.5 秒 ± 247 毫秒（平均 ± 标准偏差，2 次运行，每次 5 次循环）

test2（使用列表）

%timeit -r 2 -n 5 teste2(resp)

每个循环 2.05 秒 ± 8.17 毫秒（平均值 ± 标准偏差，2 次运行，每次 5 次循环）

test3（使用 import pandasticsearch as pdes）

%timeit -r 2 -n 5 teste3(resp)

每个循环 39.2 毫秒 ± 5.89 毫秒（平均 ± 标准偏差，2 次运行，每次 5 次循环）

test4（使用 from pandas.io.json import json_normalize）

%timeit -r 2 -n 5 teste4(resp)

每个循环 387 毫秒 ± 19 毫秒（平均值 ± 标准偏差，2 次运行，每次 5 次循环）

希望对大家有用

代码：

index = 'teste_85'
    size = 10000
    fields = True
    sort = ['col1','desc']
    query = 'teste'
    range_gte = '2016-01-01'
    range_lte = 'now'
    resp = esc.search(index = index,
                        size = size,
                        scroll = '2m',
                        _source = fields,
                        doc_type = '_doc',
                        body = {
                            "sort" : { "{0}".format(sort[0]) : {"order" : "{0}".format(sort[1])}},
                            "query": {
                                    "bool": {
                                    "must": [
                                        { "query_string": { "query": "{0}".format(query) } },
                                        { "range": { "anomes": { "gte": "{0}".format(range_gte), "lte": "{0}".format(range_lte) } } },
                                    ]
                                    }
                                }
                                })

    def teste1(resp):
        df = pd.DataFrame(columns=list(resp['hits']['hits'][0]['_source'].keys()))
        for hit in resp['hits']['hits']:
            df = df.append(df.from_dict(hit['_source'], orient='index').T)
        return df

    def teste2(resp):
        col=list(resp['hits']['hits'][0]['_source'].keys())
        for hit in resp['hits']['hits']:
            df = pd.DataFrame(list(hit['_source'].values()), col).T
        return df

    def teste3(resp):
        df = pdes.Select.from_dict(resp).to_pandas()
        return df

    def teste4(resp):
        df = json_normalize(resp['hits']['hits'])
        return df

【讨论】：

在 test2 中，返回的 df 将仅包含 resp['hits']['hits'] 的最后一个元素，你没有附加它吗？

【解决方案5】：

如果您的请求可能从 Elasticsearch 返回超过 10,000 个文档，则需要使用 Elasticsearch 的滚动功能。这个函数的文档和示例很难找到，所以我将为您提供一个完整的工作示例：

import pandas as pd
from elasticsearch import Elasticsearch
import elasticsearch.helpers


es = Elasticsearch('http://localhost:9200')

body={"query": {"match_all": {}}}
results = elasticsearch.helpers.scan(es, query=body, index="my_index")
df = pd.DataFrame.from_dict([document['_source'] for document in results])

只需编辑以“my_”开头的字段以对应您自己的值

【讨论】：

【解决方案6】：

对于遇到此问题的任何人.. @CT Zhu 有一个很好的答案，~~但我认为它有点过时了。~~ 但是当您使用 elasticsearch_dsl 包时。结果有点不同。在这种情况下试试这个：

# Obtain the results..
res = es_dsl.Search(using=con, index='_all')
res_content = res[0:100].execute()
# convert it to a list of dicts, by using the .to_dict() function
res_filtered = [x['_source'].to_dict() for x in res_content['hits']['hits']]

# Pass this on to the 'from_dict' function
A = pd.DataFrame.from_dict(res_filtered)

【讨论】：

【解决方案7】：

这里有一些您可能会发现对您的工作有用的代码。它简单且可扩展，但在面对仅从 ElasticSearch 中“抓取”一些数据进行分析时为我节省了大量时间。

如果您只想获取本地主机的给定索引和 doc_type 的所有数据，您可以这样做：

df = ElasticCom(index='index', doc_type='doc_type').search_and_export_to_df()

您可以使用通常在 elasticsearch.search() 中使用的任何参数，或指定不同的主机。您还可以选择是否包含 _id，并指定数据是在“_source”还是“fields”中（它会尝试猜测）。它还尝试默认转换字段值（但您可以将其关闭）。

代码如下：

from elasticsearch import Elasticsearch
import pandas as pd


class ElasticCom(object):

    def __init__(self, index, doc_type, hosts='localhost:9200', **kwargs):
        self.index = index
        self.doc_type = doc_type
        self.es = Elasticsearch(hosts=hosts, **kwargs)

    def search_and_export_to_dict(self, *args, **kwargs):
        _id = kwargs.pop('_id', True)
        data_key = kwargs.pop('data_key', kwargs.get('fields')) or '_source'
        kwargs = dict({'index': self.index, 'doc_type': self.doc_type}, **kwargs)
        if kwargs.get('size', None) is None:
            kwargs['size'] = 1
            t = self.es.search(*args, **kwargs)
            kwargs['size'] = t['hits']['total']

        return get_search_hits(self.es.search(*args, **kwargs), _id=_id, data_key=data_key)

    def search_and_export_to_df(self, *args, **kwargs):
        convert_numeric = kwargs.pop('convert_numeric', True)
        convert_dates = kwargs.pop('convert_dates', 'coerce')
        df = pd.DataFrame(self.search_and_export_to_dict(*args, **kwargs))
        if convert_numeric:
            df = df.convert_objects(convert_numeric=convert_numeric, copy=True)
        if convert_dates:
            df = df.convert_objects(convert_dates=convert_dates, copy=True)
        return df

def get_search_hits(es_response, _id=True, data_key=None):
    response_hits = es_response['hits']['hits']
    if len(response_hits) > 0:
        if data_key is None:
            for hit in response_hits:
                if '_source' in hit.keys():
                    data_key = '_source'
                    break
                elif 'fields' in hit.keys():
                    data_key = 'fields'
                    break
            if data_key is None:
                raise ValueError("Neither _source nor fields were in response hits")

        if _id is False:
            return [x.get(data_key, None) for x in response_hits]
        else:
            return [dict(_id=x['_id'], **x.get(data_key, {})) for x in response_hits]
    else:
        return []

【讨论】：

【解决方案8】：

使用elasticsearch_dsl，您可以搜索文档，通过 id 获取它们等。

例如

from elasticsearch_dsl import Document

# retrieve document whose _id is in the list of ids
s = Document.mget(ids,using=es_connection,index=myindex)

或

from elasticsearch_dsl import Search

# get (up to) 100 documents from a given index
s = Search(using=es_connection,index=myindex).extra(size=100)

然后，如果您想创建 DataFrame 并在数据帧索引中使用 elasticsearch id，您可以执行以下操作：

df = pd.DataFrame([{'id':r.meta.id, **r.to_dict()} 
                            for r 
                            in s.execute()]).set_index('id',drop=True)

【讨论】：

【解决方案9】：

redata = map(lambda x:x['_source'], res['hits']['hits'])
pd.DataFrame(redata)

如果我只使用 pandas 模块，那将是最好的解决方案。就我而言，这些代码花费了 20.5 毫秒

如果使用pandas.io.json.json_normalize(res['hits']['hits'])，会花费291ms，结果不一样。

【讨论】：