[ELK|Python] REST API로 ES에 데이터 색인하기

개요

열심히 구현해둔 Whisper 배치 시스템을 통해 나온 STT Text 데이터를 통해 의미있는 서비스를 만드는 것이 이번 과제였다. 
그 중 TextRank 알고리즘을 이용한 문서 요약 과제가 추가되었으나 TextRank 알고리즘은 이론적으로 설명 가능한 대안이나 현실적이진 못했다..
STT 데이터 자체는 애초에 정확하지 않다. 그래서 STT Text 를 사용하여 내용, 키워드를 파악해볼 수 있도록 ELK로 데이터 분류 작업을 해보기로 하였다. (블로그 -ES로 간편해진 텍스트 분류)

---

ElasticSearch가 검색하는 방법

우선, ElasticSearch와 RDBMS(Mysql, Oracle)의 차이에 대해 알아보자. 아래는 가장 큰 두가지 요소인 인덱싱과 검색방법의 차이이다.

• INDEX
• ElasticSearch의 인덱스는 문서형태로 특정 열이 아닌 전체 문서를 인덱싱-역인덱싱 형태로 검색한다면, RDBMS는 주로 B-트리 형태를 사용하여, 테이블의 특정 열을 기반으로 검색한다.
• ElasticSearch의 키워드 검색과 RDBMS의 LIKE 검색
• RDBMS의 LIKE 검색은 텍스트 일치를 위해 모든 레코드를 순차적으로 스캔하지만, ElasticSearch는 역색인 방식을 사용해 대용량 데이터에서 더 빠른 조회를 가능하게 한다.
• 또한 LIKE는 정해진 규격의 검색('나무', '%나무', '%나무%' 등) 만이 가능하나 ElasticSearch는 여러 조건을 걸어 검색할 수 있다.('나무', '나뭇', '나뭇잎' 등)

이런 특징으로 인해 ES는 실시간 검색 및 분석에 유리하고, Document 기반 검색 엔진을 구현할 때 많이 사용한다.
역인덱싱
이 중 가장 흥미있었던 역인덱싱에 대해 더 자세히 알아보았다. ES에서는 데이터를 저장한다는 표현보다 색인한다는 표현을 더 자주 쓴다고 한다. 데이터를 저장할 때마다 역인덱싱 테이블을 만드는 것이 이 이유이다. 이 부분이 신기했다. 어떻게 데이터를 저장할 때마다 역인덱싱 테이블을 만들지? 이미 앞에 인덱스가 존재하는지 확인해봐야 할 건데 그것도 시간이 많이 걸리지 않을까,, 그리고 '안녕하세요', '안녕하십니까' 라는 단어들은 '안녕'을 검색해도 둘 다 조회되어야 할텐데,,?? 그걸 다 저장한다고?

알고보니 이런 과정들을 거친다고 한다. '캐릭터 필터'에서 필요없는 단어들(불용어)을 제거해주고, 토크나이저에서는 일정 기준에 따라 문장 속 단어들을 Term이라고 한다. 이 Term을 기준으로 한 테이블 형식을 만들어준다. 토큰 필터에서는 해당 term들을 이용해 이 테이블을 가공한다. '안녕하세요', '안녕하십니까'를 묶어서 '안녕'으로 처리해주기도 하고(접미사나 어미를 뺀다), 또는 이 둘을 각 테이블로 넣되 동의어로 간주하여 Synonym Token Filter을 사용해 처리하기도 한다. 
그렇다면 내 시스템에 적용해봐도 좋겠다 ! 

---

RDB(Oracle)에서 데이터 가져와 ElasticSearch 에 적재

우선 ElasticSearch에 데이터가 적재되어야 하기 때문에 운영 중인 RDBMS에 생성된 데이터를 동기화 시켜주어야 한다. 이 때 Logstash와 JDBC를 사용하여 적재하는 방법이 있으나 데이터를 일부분 가공해야 했기 때문에 Logstash 방식을 사용하지 않고(굳이 데이터를 수집하지 않고) jupyter notebook 내에서 Oracle 데이터를 불러온 후 JSON 형태의 REST API로 ES에 전송하는 방식을 택했다. ( ES의 편리한 점은 "자체 restAPI" 를 지원한다는 점에 있다. 👍 )
프로세스
1. Oracle에서 RECRD_ID, RECRD_TEXT  형태의 데이터 가져오기
2. 가져온 데이터를 적절히 가공 ( whisper 데이터 시간 text는 제외시킨다 ) 
3. 해당 결과를 Elasticsearch로 출력 by REST API ( @timestamp 필드 추가시킨다 )
4. Kibana에서 결과에 대한 시각화
5. Index pattern 생성하여 데이터 스트림 형태로 저장 ( 이 때 3일치 것만 저장하고 이전 데이터는 삭제한다 ) 

---

REST API로 ES에 데이터 적재하기 - PUT, POST

ES에 파이썬으로 적재하기 위해 ES 문서나 파이썬 문서를 참고하였다.
아래 문서를 참고해 간단히 알아보자. ES에 document(데이터)를 색인하기 위해서는 index 이름, id, document의 정보를 담아서 전송하면 된다. 해당 형식은 PUT 방식으로 특징 ID(1)를 가지는 index에 데이터를 입력하는 예제이다.

이 때 ID를 굳이 지정해주지 않으면 ID를 자동으로 생성해주는 POST 형태에 해당한다. <인덱스명>/_doc까지만 입력하게 되면 자동으로 임의의 id가 생성된다. ( 예를 들면 이런 형식 - "_id" : "ZuFv12wBspWtEG13dOut" ).

_bulk API 를 사용한 대량 데이터 로드

일단 나는 (RECRD_ID, RECRD_TEXT) 로 이루어진 다량의 데이터를 넣어야 했기 때문에 대량 문서 색인 방식인 _bulk API를 사용해 JSON 형태의 document를 한 번에 전송하였다. (참고, 참고)
_bulk API는 요청 시 한 번의 API 요청으로 대량의 데이터를 보낼 수 있고, 해당 JSON을 이용해 병렬 및 일괄 처리를 하기 때문에 매우 효율적이다.
아래와 같이 간단히 구현할 수 있었다.

import requests
from elasticsearch import Elasticsearch, helpers
from elasticsearch import RequestsHttpConnection
from datetime import datetime, timedelta  

es = Elasticsearch(['https://{ES URL}:{ES PORT}'], connection_class=RequestsHttpConnection,
                   use_ssl=True, verify_certs=False, http_auth=('{ES_ID}', '{ES_PW}'), timeout=300)

index_name = "whisper_test"

bulk_data = []

for recrd in query_result: #recrd = (RECRD_ID, RECRD_TEXT)
    recrd_id, text = recrd
    
    yesterday = datetime.utcnow() - timedelta(days=1) #UTC 시간 형태로 어제 날짜를 가져온다.
    timestamp = yesterday.isoformat()
    
    doc = {
        "id": recrd_id,
        "text": text,
        "@timestamp": timestamp
    }
    
    action = {
            "_index": index_name,
            "source": doc
    }
    bulk_data.append(action)
        
helpers.bulk(es, bulk_data) #한 번의 API로 bulk 연산 실행

Dev Tool을 이용해 조회해보면 잘 색인되는 것을 볼 수 있다. (아래는 참고 이미지)

GET /whisper_test/_search?pretty

다음에는 적재된 ES 데이터와 역인덱싱 테이블을 이용해 Kibana에 데이터를 조회해 보자 !