从ElasticSearch结果创建DataFrame

或者您可以使用pandas的json_normalize函数:

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

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

我测试了所有性能答案，然后发现该pandasticsearch方法是最快的方法：

测试：

test1（使用from_dict）

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

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

test2（使用列表）

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

每个循环2.05 s±8.17 ms（平均±标准偏差，两次运行，每个循环5次）

test3（使用import pandasticsearch作为pdes）

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

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

test4（从pandas.io.json导入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

更好的是,你可以使用梦幻般的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()

有一个很好的玩具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,将列表中的每个项目(即a dictionary)读入aDataFrame

2,我们可以将列表中的所有项目DataFrame按concat行排列,因为我们将为每个项目执行步骤#1,我们可以使用map它来执行此操作.

3,然后我们访问标有的列 'fields'

4,我们可能想要旋转DataFrame90度(转置),reset_index如果我们想要索引是默认int序列.