[FastAPI] ElasticSearch를 활용한 검색 속도 향상

ElasticSearch를 활용한 검색 속도 향상 왜 궁금했을까❓

이번 포스팅에서는 ElasticSearch를 프로젝트에 적용시키고 MySQL을 활용해서 구현했던 뉴스 검색 기능과 성능 차이를 비교해보려고 한다.

ElasticSearch의 개념과 작동 원리가 궁금하다면 아래 포스팅을 통해서 확인할 수 있다.

[FastAPI] ElasticSearch

 

1. Docker / Elasticsearch & Kibana 설치

services:
  elastic:
    container_name: elastic
    image: docker.elastic.co/elasticsearch/elasticsearch:7.10.0
    restart: always
    ports:
      - 9200:9200
    environment:
      - discovery.type=single-node

  kibana:
    container_name: kibana
    image: docker.elastic.co/kibana/kibana:7.10.1
    restart: always
    ports:
      - 5601:5601
    environment:
      ELASTICSEARCH_URL: http://elastic:9200
      ELASTICSEARCH_HOSTS: http://elastic:9200
      privileged: true

• ElasticSearch를 다중 노드로 구성하여 안정성과 가용성을 확보해야 하지만 짧은 프로젝트 기간을 고려하여 복잡성이 증가할 것이라 예상되어 싱글 노드로 구성했다.
• Kibana와 Elasticsearch 버전이 맞아야 서로 호환이 되어 작동한다.

2. FastAPI

2.1. Elasticsearch 설치

pip install elasticsearch==7.10.0

2.2. Elasticsearch 초기 설정

@app.on_event("startup")
def startup():
    es = Elasticsearch('localhost:9200')
    index_name = 'article'

    with open('./mapping.json', 'r') as f:
        mapping = json.load(f)

    if es.indices.exists(index=index_name):
        es.indices.delete(index=index_name, ignore=400)
    es.indices.create(index=index_name, body=mapping, ignore=400)

    print("completed es_init")

• Elasticsearch의 주소를 입력하여 연결한다.
• 인덱스 이름을 'article'로 지정하고 mapping 정보를 가져온다.
• 만약, 인덱스가 존재한다면 삭제하고 없다면 mapping 정보를 이용하여 인덱스를 생성한다.

{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 1
  },
  "mappings": {
    "properties": {
      "article_id": {
        "type": "long"
      },
      "title": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "content": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 10000
          }
        }
      },
      "image_url": {
        "type": "keyword"
      }
    }
  }
}

• 샤드의 경우 샤드당 약 50GB의 데이터가 적재되는 것을 권장한다고 한다. 뉴스타 서비스의 경우 사이드 프로젝트와 같은 개념이라서 1개의 샤드를 두고 노드의 부하를 줄이기로 결정했다.
• 레플리카 샤드의 경우도 1개를 두고 혹시 모를 장애 상황에 대비하고 읽기 작업을 분산 처리하여 성능을 높여주었다.
• 문자열 타입의 경우 text와 keyword가 존재하는데 text의 경우 형태소 분석이 이루어져 분석기에 의해 문장이 분리가 되어 색인된다. 하지만, keyword의 경우 분석기를 거치지 않기 때문에 문장 전체가 색인된다. 검색의 정확도를 높이기 위해 keyword를 채택하였다.

2.3. 뉴스 데이터 초기 설정 및 업데이트

@router.get("/crawling")
def start_crawling():
  ...
  article_id = es_service.last_article_id()
  
  if article_id.loc[0]['article_id'] == 0:
    es_service.init_es()
  else:
    es_service.add_es(article_id.loc[0]['article_id'])

  es_service.update_last_article_id()
  ...

• MySQL DB에는 last_article_id라는 테이블에 ES에 마지막으로 들어간 뉴스의 ID 값을 담고 있다.
• 해당 값이 0이라면 아직 ES가 초기 설정이 되지 않은 것이므로 ES를 초기 설정을 한다.
• 0이 아니라면 ES에 마지막 ID 값부터 크롤링한 가장 최신의 뉴스의 ID까지의 데이터를 업데이트한다.
• 그리고 last_article_id를 업데이트하여 마지막 뉴스 ID를 갱신한다.

def last_article_id(db: Session = Depends(get_db)):
    article_id = es_crud.get_last_article_id(db)
    return article_id

def get_last_article_id(db: Session):
    article_id = pd.read_sql("SELECT * FROM last_article_id", con = engine)
    return article_id
    
def init_es(db: Session = Depends(get_db)):
    articles = es_crud.get_article_all(db)
    es = Elasticsearch(f"{ES['host']}")

    for idx, row in articles.iterrows():
        doc = {'article_id' : row['article_id'], 'title' : row['title'], 'content' : row['content'], 'image_url' : row['image_url']}
        es.index(index='article', doc_type='_doc', body=doc)
        es.indices.refresh(index='article')

def get_article_all(db: Session):
    articles = pd.read_sql("SELECT article_id, title, content, image_url FROM article", con = engine)
    return articles
    
def add_es(article_id : int, db: Session = Depends(get_db())):
    articles = es_crud.get_recent_article(article_id, db)
    es = Elasticsearch(f"{ES['host']}")

    for idx, row in articles.iterrows():
        doc = {'article_id' : row['article_id'], 'title' : row['title'], 'content' : row['content'], 'image_url' : row['image_url']}
        es.index(index='article', doc_type='_doc', body=doc)
        es.indices.refresh(index='article')

def get_recent_article(last: int, db: Session):
    query = f"SELECT * FROM article WHERE article_id > {last}"
    articles = pd.read_sql(query, con=engine)
    return articles

def update_last_article_id(db: Session = Depends(get_db)):
    max = es_crud.get_max_article_id(db)
    max_id = max.loc[0]['max(article_id)']
    es_crud.update_last_article_id(max_id, db)

def get_max_article_id(db: Session):
    max = pd.read_sql("SELECT max(article_id) FROM article", con = engine)
    return max

 

2.4. 뉴스 검색 API

@router.post("", response_model=List[Articles], dependencies=[verify_header()])
def search(keyword: Keyword):
    keyword = keyword.keyword

    res = []
    index_name = 'article'

    es = Elasticsearch(f"{ES['host']}")
    max_article_id = es.search(
        index=index_name,
        body={
            'size': 1,
            'query': {'match_all': {}},  # 모든 문서를 검색
            'sort': [{'article_id': {'order': 'desc'}}]  # article_id를 내림차순으로 정렬
        }
    )
    cur = 0

    for m in max_article_id['hits']['hits']:
        cur = m['_source']['article_id']

    results = es.search(
        index=index_name,
        body={
            'from': 0,
            'size': 10,
            'query': {
                'bool': {
                    'must': [{'match': {'content.keyword': keyword}}],
                    'filter': [{'range': {'article_id': {'gte': cur/2}}}]
                }
            },
            'collapse': {'field': 'title'}
        }
    )
    for result in results['hits']['hits']:
        article_id = result['_source']['article_id']
        title = result['_source']['title']
        image_url = result['_source']['image_url']
        article = Articles(article_id=article_id, title=title, image_url=image_url)
        res.append(article)
    return res

• 일단 ES에 있는 뉴스 데이터의 가장 최신 뉴스의 ID를 가져와서 cur 변수에 저장한다.
• 10개의 검색 결과를 가져오며 뉴스 전문에 사용자가 입력한 keyword가 무조건 포함된 기사를 가져온다.
• gte 옵션을 통해서 cur / 2 > article_id를 만족하는 기사를 검색하여 최대한 최신 기사를 가져오도록 한다.
• collapse 옵션을 통해 제목이 겹칠 경우 중복을 제거하여 검색한다.

3. 검색 정확도 향상

3.1. Nori 형태소 분석기

• Elasticsearch에서 한국어를 검색할 때는 한글 형태소 분석기인 Nori를 사용한다.
• 기본으로 제공되는 Standard Analyzer를 사용하면 White Space를 기준으로 Tokenize 되긴하지마 한국어의 원형을 찾지 못해 정확도가 떨어진다.

3.2. Nori Analyzer 설치

6.7.2 노리 (nori) 한글 형태소 분석기 | Elastic 가이드북

sudo bin/elasticsearch-plugin install analysis-nori

• Elasticsearch 홈 디렉토리에서 위 명령어를 통해 Nori Plugin 설치

elasticsearch-plugin list

• 위 명령어를 통해 설치 여부 확인하고 재시작

3.3. Nori Analyzer 설정

{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 1,
    "analysis": {
      "analyzer": {
        "nori": {
          "type": "custom",
          "tokenizer": "nori_tokenizer",
          "decompound_mode": "mixed"
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "article_id": {
        "type": "long"
      },
      "title": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "content": {
        "type": "text",
        "fields": {
          "nori": {
            "type": "text",
            "analyzer": "nori"
          }
        }
      },
      "image_url": {
        "type": "keyword"
      }
    }
  }
}

• decompound_mode를 mixed로 설정하여 어근과 합성어를 모두 저장하여 검색의 질을 높였다.

results = es.search(
  index=index_name,
  body={
    'from': 0,
    'size': 10,
    'query': {
      'bool': {
        'must': [{'match': {'content.nori': keyword}}],
        'filter': [{'range': {'article_id': {'gte': cur/2}}}]
      }
    },
  'collapse': {'field': 'title'}
  }
)

• content.nori로 뉴스 검색 API 수정

4. 성능 비교

약 20,000개의 뉴스 데이터를 가지고 있을 때 관련이 없는 키워드를 넣어 full scan을 시도해 보았다.

4.1. MySQL LIKE 함수

• POSTMAN을 이용하여 MySQL LIKE로 구현된 검색 API에 요청한 결과 387ms의 성능을 보였다.

4.2. Elasticsearch

• 동일한 키워드로 POSTMAN을 이용하여 Elasticsearch로 구현된 검색 API에 요청한 결과 26ms의 성능을 보였다.

네트워크나 서버의 성능이 영향이 있을 수도 있지만 확연한 차이를 보이며 검색 시간이 단축된 것을 알 수 있다.