从ElasticSearch结果创建DataFrame

Question

我正在尝试使用ElasticSearch的一个非常基本的查询结果在pandas中构建一个DataFrame.我得到了我需要的数据,但它的结果是以一种方式构建正确的数据框.我真的只关心每个结果的时间戳和路径.我尝试了一些不同的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'}

我不能为我的生活弄清楚谁将这个结果,放到熊猫的数据框中.所以对于我返回的2个结果,我希望有一个数据帧.

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

Answer 1

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

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

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

Answer 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()

Answer 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,将列表中的每个项目(即a dictionary)读入aDataFrame

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

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

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

Answer 4

如果您的请求可能从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_”开头的字段以对应于您自己的值

Answer 5

我测试了所有性能答案，然后发现该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