Создание DataFrame из результатов ElasticSearch

Question

Я пытаюсь создать DataFrame в pandas, используя результаты очень простого запроса к 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'} 

Я не могу за жизнь мне понять, кто, чтобы получить этот результат, в dataframe в панд. Таким образом, для двух результатов, которые я вернул, я ожидал бы, например, dataframe.

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

Answer 1

Существует хорошая игрушка называется 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, мы можем поставить все элементы в списке в большой DataFrame по concat их по-разному, так как мы будем делать шаг №1 для каждого элемента, мы можем использовать map, чтобы сделать это.

3, Тогда доступ к столбцам, меченные 'fields'

4, мы, вероятно, хотите, чтобы повернуть DataFrame 90 градусов (транспонирование) и reset_index, если мы хотим, чтобы индекс будет int последовательность по умолчанию.

Answer 2

Вот немного кода, который вы могли бы оказаться полезными для вашей работы. Он прост и расширяем, но сэкономил мне много времени, когда вы столкнулись с «захватом» некоторых данных из 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 [] 

Answer 3

Или вы можете использовать функцию json_normalize панд:

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

А затем фильтровать результат dataframe по именам столбцов

Answer 4

еще лучше, вы можете используйте фантастическую библиотеку 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()