Elasticsearch x Python - 엘리스틱 서치 연동

Elasticsearch가 설치되어 있는 환경에서, Python으로 간단한 질의를 해 본 스크립트이다.

 

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import json
from  elasticsearch import Elasticsearch

host = '150.19.5.30'
port = '9200'
target = 'http://{}:{}'.format(host,port)
#print(target)

es = Elasticsearch( [{'host':host, 'port':port}])
docs = es.search(index='skt_app_vdi_file-20171123', filter_path=['hits.hits._*'])

#len(docs['hits']['hits'])
docs['hits']['hits'][1]['_source']

 

 

Elasticseach 라이브러리를 import 하고, es를 할당하여 index 및 document를 얻어 온다.
document를 가져오면, JSON형태가 상당히 번거로운 구조를 가지고 있다.

따라서, 원하는 본문을 얻기위해서, docs['hits']['hits'] 후 부터 본격적인 검색 결과라 하겠다.

hits는 elastic에서 Keyword 검색에 유사성을 나타내고 있는 score 값이다.

 

2. 클래스로 만들어 보자

기왕 Python 으로 엘라스틱을 사용해본 김에 클래스로 만들어 보자. 

(사실 그냥 JSON형태로 묶어서 rest 로 날려주는게 더 나을 지도 모른다.)

 

class MyElastic():
    def __init__(self, index, hosts, ports, size=1000) :
        self.index = index
        self.host = hosts
        self.ports = ports
        self.body = ""
        self.aggs = None
        self.res = None
        self.total = 0
        self.df = None
        self.s = None
        self.cols = None
        self.q = None
        
        server_list = [ {'host':x, 'port':y}  for x,y in zip(hosts,ports) ]
        self.es = Elasticsearch( server_list, size=size )        
        self.s = Search(index = index).using(self.es)
           
    def search(self, q, nSize=1000, iFrom=0):
        s = self.s
        s = s.query(q)
        
        s = s[iFrom : iFrom + nSize]
        res = s.execute()
        
        self.total = res.hits.total        
        self.q, self.s, self.res = q,s, res
    
    def search_wild(self, f="emp_id", v="*", nSize =1000 ):
        s = self.s       
        q_dict = {"query":{"wildcard":{str(f)+".keyword": v}}, "size": nSize }
        s = s.query(Q("match_all"))
        s = s.update_from_dict(q_dict)
        res = s.execute()
        
        self.total = res.hits.total        
        self.s, self.res = s, res    
       
        
    def search_from_body(self, body):
        return None
    
    def get_by_id(self, id, s_type="match"):
        q = Q(s_type, emp_id=id)
        self.search(q)
        self.df = self.to_df(self.res)
        return self.df
        
        
    def to_df(self, res):
        aDict = [ x['_source'] for x in res.hits.hits]
        df = pd.DataFrame(aDict)        
        return df
    
    def get_cols(self,res):
        if self.total!=0:
            cols = list(res.hits.hits[0]['_source'].keys())
            return cols
        else:
            return None

Pandas로 엘라스틱 서치 검색결과를 연동해 서 사용해 볼 수 있다. 

import pandas as pd

arr_dict = [ x['_source'] for x in docs['hits']['hits'] ]
df_raw = pd.DataFrame(arr_dict)
print("Shape:", df_raw.shape)
print("columns:", df_raw.columns)
df_raw.head(3)

plt.rcParams["font.family"] = 'New Gulim'   # 한글 폰트 지정
g = sns.countplot(data=df_raw, x='request')
g.set_xticklabels(g.get_xticklabels(), rotation = 30)
plt.show()

df_g = df_raw.groupby(['client_ip','portdesc'])['emp_id'].count()
hmap = df_g.unstack(['client_ip']).fillna(0).astype('int64')

fig, ax = plt.subplots(figsize=(15,10))
g = sns.heatmap(hmap, annot=True, fmt='d', linewidths=.5, vmin=0, vmax=100, ax=ax)
g.set_xticklabels(g.get_xticklabels(), rotation = 30)
plt.show()