대용량 텍스트 검색 및 전송 최적화 엔진 (4) Elasticsearch 기반 풀텍스트 검색 (上)

Elasticsearch 인프라 구성

🤖 AI AGENT | Elasticsearch 메이저 버전 선택 가이드

Docker Hub의 최신 태그 현황을 확인한 결과, 8.x 대의 최신 안정 버전인 8.19.16과 최신 메이저 버전인 9.4.2가 모두 릴리즈되어 있습니다.
현재 Python 3.12 환경에서 9.x 버전의 기술적 제약(클라이언트 지원, xpack 보안 비활성화, Nori 플러그인 호환 등)은 완벽히 해결된 상태입니다. 따라서 기술적인 장벽은 없으며, 사용자의 학습 목적과 성향에 따라 버전을 선택하실 수 있습니다.
아래의 비교 분석을 확인하신 후 원하시는 방향을 선택해 주세요.

---

1. 버전별 특징 및 장단점 비교

🔹 8.x 버전 (8.19.16) — 안정성과 풍부한 레퍼런스 (권장)

• 압도적인 문제 해결 자료: 대다수의 FastAPI 연동 예제, StackOverflow 답변, 튜토리얼이 8.x 기준이므로 에러 발생 시 검색 및 트러블슈팅이 매우 수월합니다.
• Breaking Changes 회피: 9.x로 넘어가면서 제거된 이전 API 스펙이나 Query DSL 문법 오류로부터 안전하므로, 기존 강의나 자료를 매끄럽게 따라갈 수 있습니다.
• 검증된 편의성: 학습 편의를 위해 SSL/인증을 끄는 설정(xpack.security.enabled=false)이 안정적으로 동작합니다.

🔸 9.x 버전 (9.4.2) — 최신 스택 경험 및 도전

• 미래 지향적 설계: 현재 시점에서 가장 최신의 아키텍처와 공식 라이브러리(elasticsearch9) 스펙을 직접 경험하고 체득할 수 있습니다.
• 잠재적 리스크: 메이저 버전 변경으로 인해 하위 호환성이 끊긴 API가 있을 수 있으며, 출시된 지 비교적 얼마 되지 않아 트러블슈팅 자료가 부족할 수 있습니다.

---

2. 최종 결론 및 제안

어느 버전을 선택하셔도 인프라 환경 구축부터 구현까지 완벽하게 지원해 드립니다. 사용자님의 선호에 따라 방향을 선택해 주세요.

• 선택 1) 안정형 ➔ 8.19.16 버전 선택 (추천)
  보안 및 버그 수정 패치가 모두 반영되면서도 하위 호환성을 유지할 수 있는 가장 최선이자 안전한 선택입니다.
• 선택 2) 도전형 ➔ 9.4.2 버전 선택
  최신 기능과 최신 클라이언트를 다뤄보며 트러블슈팅을 직접 경험해보고 싶을 때 선택합니다.

그래도 AI가 삽질을 좀 줄여주지 않을까 하여 9.4.2 버전을 사용해 보기로 했다.

먼저, 한글 형태소 분석기 Nori 플러그인을 포함한
커스텀 Elasticsearch 이미지를 빌드하기 위한 도커 파일을 생성한다.

infra/elasticsearch.Dockerfile

FROM docker.elastic.co/elasticsearch/elasticsearch:9.4.2

RUN bin/elasticsearch-plugin install --batch analysis-nori

이 Elasticsearch 이미지와
색인 현황을 시각화 해주는 Kibana 이미지를 띄우는 파일을 작성한다.

infra/compose.yml

services:
  elasticsearch:
    build:
      context: .
      dockerfile: elasticsearch.Dockerfile
    container_name: elasticsearch
    environment:
      - discovery.type=single-node # 단일 노드 모드로 실행하여 클러스터 노드 탐색을 생략
      - xpack.security.enabled=false # X-Pack 보안 기능 비활성화로 로컬 개발 편의성 확보
      - xpack.security.transport.ssl.enabled=false # 노드 간 통신용 SSL/TLS 암호화 비활성화
      - xpack.security.http.ssl.enabled=false # HTTP API 통신 시 HTTPS 대신 인증서 설정 불필요한 HTTP 통신 사용
      - ES_JAVA_OPTS=-Xms512m -Xmx512m # JVM 힙 메모리 크기를 최소/최대 512MB로 제한하여 로컬 시스템의 과도한 메모리 사용 방지
    ports:
      - "9200:9200"
    volumes:
      - es_data:/usr/share/elasticsearch/data
  kibana:
    image: docker.elastic.co/kibana/kibana:9.4.2
    container_name: kibana
    environment:
      # Kibana가 연결할 Elasticsearch 주소 설정 (Docker 네트워크 내부의 서비스 이름으로 지정)
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    ports:
      - "5601:5601"
    volumes:
      - kibana_data:/usr/share/kibana/data
    # Elasticsearch가 먼저 정상 기동된 후 실행되도록 의존성 명시
    depends_on:
      - elasticsearch
volumes:
  es_data:
  kibana_data:

이제 컨테이너를 빌드하고 데몬으로 실행한다.

~/workspace/fast-text-search/infra$ docker compose up -d --build

작동 테스트를 해보면,

~$ curl -X GET "http://localhost:9200/"

{
  "name" : "3774329fe3c4",
  "cluster_name" : "docker-cluster",
  "cluster_uuid" : "NzBjN_L4RKCf5mKeY1MMKA",
  "version" : {
    "number" : "9.4.2",
    "build_flavor" : "default",
    "build_type" : "docker",
    "build_hash" : "c402c2b36d90eae29c0182f86bd9050fd0b746cc",
    "build_date" : "2026-05-25T22:10:36.017759931Z",
    "build_snapshot" : false,
    "lucene_version" : "10.4.0",
    "minimum_wire_compatibility_version" : "8.19.0",
    "minimum_index_compatibility_version" : "8.0.0"
  },
  "tagline" : "You Know, for Search"
}

~$ curl -X GET "http://localhost:9200/_cat/plugins?v"

name         component     version
3774329fe3c4 analysis-nori 9.4.2

게이트웨이 환경설정 및 의존성 추가

Python

FastAPI 백엔드가 Elasticsearch 9.x 버전과 비동기로 연동할 수 있도록
Python 개발 환경을 세팅한다.

먼저 pyproject.toml 에 Elasticsearch 의존성을 추가한다.

gateway/pyproject.toml

[project]
name = "gateway"
version = "0.1.0"
description = "대용량 텍스트 검색 및 최적화 시스템의 FastAPI 게이트웨이"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "brotli-asgi>=1.6.0",
    "fastapi>=0.137.0",
    "pydantic-settings>=2.14.1",
    "uvicorn[standard]>=0.49.0",
    "fast-text-engine>=0.1.0",
    "elasticsearch>=9.0.0,<10.0.0",
    "aiohttp>=3.9.0",
]

[dependency-groups]
dev = [
    "httpx2>=2.4.0",
    "mypy>=2.1.0",
    "ruff>=0.15.17",
    "maturin>=1.14.0",
    "pytest>=9.1.0",
]

[tool.pytest.ini_options]
pythonpath = ["."]

[tool.uv.sources]
fast-text-engine = { path = "../engine", editable = true }

~/workspace/fast-text-search/gateway$ uv sync

FastAPI 앱에서 연동할 인덱스명을 동적으로 지정할 수 있도록 설정 파일을 수정하고
.env 파일에 적절한 값을 작성한다.
실무에서는 환경 변수 파일을 외부에 공유하지 않도록 유의한다.

gateway/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    """API 게이트웨이의 환경변수 및 설정을 로드하고 관리하는 클래스."""
    GATEWAY_HOST: str
    GATEWAY_PORT: int
    CORS_ORIGINS: str
    ELASTICSEARCH_URL: str
    ELASTICSEARCH_INDEX: str

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore"
    )

settings = Settings()

.env

GATEWAY_HOST=0.0.0.0
GATEWAY_PORT=8000
CORS_ORIGINS=http://localhost:5173
ELASTICSEARCH_URL=http://localhost:9200
ELASTICSEARCH_INDEX=fast-text-index

schemas.py 에 Elasticsearch 관련 기능의 입출력 데이터를 정의할
Pydantic 스키마를 작성한다.
새로 추가된 스키마 외에도 HealthResponse() 에 한 줄 추가되었음을 유의하자.

gateway/schemas.py

from pydantic import BaseModel

class HealthResponse(BaseModel):
    """API 게이트웨이 및 Rust 엔진의 헬스 상태 응답 스키마."""
    status: str
    engine: str
    elasticsearch: str = "unchecked"

class TextDataResponse(BaseModel):
    """대용량 텍스트 생성 결과를 반환하는 응답 스키마."""
    lines: int
    data: list[str]

class IndexRequest(BaseModel):
    """Elasticsearch에 색인할 텍스트 라인 수를 요청하는 스키마."""
    lines: int = 10000

class IndexResponse(BaseModel):
    """색인 완료 결과를 나타내는 응답 스키마."""
    status: str
    indexed_count: int

class SearchHit(BaseModel):
    """단일 검색 매칭 정보 스키마."""
    id: str
    text: str
    highlight: list[str] = []

class SearchResponse(BaseModel):
    """풀텍스트 검색 결과 전체를 담은 응답 스키마."""
    total: int
    hits: list[SearchHit]

class QueryRequest(BaseModel):
    """검색어 및 페이징 요청 스키마."""
    q: str
    from_idx: int = 0
    size: int = 10

class IndexStatsResponse(BaseModel):
    """인덱스 통계 정보 스키마."""
    document_count: int
    segment_count: int

변경된 헬스체크 스키마대로 lasticsearch 서버 상태까지 복합적으로 점검할 수 있도록 하고
비동기 Elasticsearch 클라이언트를 안전하게 연결 및 종료하도록 설정하겠다.

gateway/app/main.py

import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from brotli_asgi import BrotliMiddleware
import hashlib

from elasticsearch import AsyncElasticsearch # NEW!

import fast_text_engine
from .config import settings
from .schemas import HealthResponse, TextDataResponse

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("gateway")

@asynccontextmanager
async def lifespan(app: FastAPI):
    """FastAPI 애플리케이션의 수명 주기를 관리하는 컨텍스트 매니저."""
    logger.info("Fast Text Search Gateway 서버 기동...")
    logger.info(f"설정 로드 완료 - Host: {settings.GATEWAY_HOST}, Port: {settings.GATEWAY_PORT}")

    # AsyncElasticsearch 비동기 커넥션 풀을 초기화하고 앱 상태에 전역 저장합니다.
    es_client = AsyncElasticsearch(settings.ELASTICSEARCH_URL)
    app.state.es_client = es_client
    logger.info(f"Elasticsearch 비동기 클라이언트 초기화 완료: {settings.ELASTICSEARCH_URL}")

    yield

    # 앱 종료 시 Elasticsearch 커넥션을 안전하게 닫아줍니다. (Graceful Shutdown)
    await app.state.es_client.close()
    logger.info("Elasticsearch 비동기 클라이언트 커넥션 닫기 완료.")
    logger.info("Fast Text Search Gateway 서버 종료...")

# (중략)
@app.get("/health", response_model=HealthResponse)
async def health_check(request: Request) -> HealthResponse:
    """게이트웨이 자체 상태와 Rust 연산 엔진의 동작 여부를 검사합니다.

    Args:
        request (Request): App state에 저장된 Elasticsearch 클라이언트에 액세스하기 위한 HTTP Request 객체.

    Returns:
        HealthResponse: 상태 검사 결과 객체.
    """

    engine_status = "error"
    try:
        engine_status = fast_text_engine.engine_health()
    except Exception as e:
        logger.error(f"Rust 엔진 헬스 체크 실패: {e}")

    es_status = "error"
    try:
        es_client: AsyncElasticsearch = request.app.state.es_client
        if await es_client.ping():
            es_status = "ok"
    except Exception as e:
        logger.error(f"Elasticsearch 헬스 체크 실패: {e}")

    return HealthResponse(
        status="ok" if (engine_status == "ok" and es_status == "ok") else "error",
        engine=engine_status,
        elasticsearch=es_status
    )

# (후략)

서버를 가동하고 테스트를 해 보면,

[터미널 A | 게이트웨이 실행]

~/workspace/fast-text-search/gateway$ uv run uvicorn app.main:app --reload

[터미널 B | 응답 확인]

~$ curl -i http://localhost:8000/health

HTTP/1.1 200 OK
date: Mon, 22 Jun 2026 04:56:13 GMT
server: uvicorn
content-length: 50
content-type: application/json

{"status":"ok","engine":"ok","elasticsearch":"ok"}%

인덱스 매핑 설계

~/workspace/fast-text-search$ mkdir scripts && touch scripts/create_index.py

scripts/create_index.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "elasticsearch>=9.0.0,<10.0.0",
#     "aiohttp>=3.9.0",
#     "python-dotenv>=1.0.0",
# ]
# ///

import os
import asyncio
import logging
from pathlib import Path
from dotenv import load_dotenv
from elasticsearch import AsyncElasticsearch

# 로그 설정
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("create_index")

env_path = Path(__file__).resolve().parent.parent / "gateway" / ".env"
load_dotenv(dotenv_path=env_path)

ELASTICSEARCH_URL = os.getenv("ELASTICSEARCH_URL")
INDEX_NAME = os.getenv("ELASTICSEARCH_INDEX")

# 인덱스 설정 및 매핑 정의
INDEX_CONFIG = {
    "settings": {
        "analysis": {
            "analyzer": {
                "nori_analyzer": {
                    "type": "custom",
                    "tokenizer": "nori_tokenizer"
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "line_num": {
                "type": "integer"
            },
            "text": {
                "type": "text",
                "analyzer": "nori_analyzer"
            }
        }
    }
}

async def main():
    # 비동기 Elasticsearch 클라이언트 연결
    es = AsyncElasticsearch(ELASTICSEARCH_URL)
   
    try:
        # 기존 인덱스가 있다면 덮어쓰기 위해 체크 및 삭제 (학습/개발 환경 전용)
        exists = await es.indices.exists(index=INDEX_NAME)
        if exists:
            logger.info(f"기존 인덱스 '{INDEX_NAME}'가 발견되어 삭제합니다.")
            await es.indices.delete(index=INDEX_NAME)
            logger.info(f"기존 인덱스 삭제 완료.")
      
        # 인덱스 생성
        logger.info(f"인덱스 '{INDEX_NAME}' 생성 중 (Nori 분석기 적용)...")
        await es.indices.create(index=INDEX_NAME, body=INDEX_CONFIG)
        logger.info(f"인덱스 '{INDEX_NAME}'가 성공적으로 생성되었습니다!")

    except Exception as e:
        logger.error(f"인덱스 생성 중 에러가 발생했습니다: {e}")
    finally:
        # 커넥션 종료
        await es.close()

if __name__ == "__main__":
    asyncio.run(main())

~/workspace/fast-text-search$ uv run scripts/create_index.py

Installed 19 packages in 11ms
2026-06-22 14:59:34,978 [INFO] HEAD http://localhost:9200/fast-text-index [status:404 duration:0.006s]
2026-06-22 14:59:34,979 [INFO] 인덱스 'fast-text-index' 생성 중 (Nori 분석기 적용)...
2026-06-22 14:59:35,057 [INFO] PUT http://localhost:9200/fast-text-index [status:200 duration:0.079s]
2026-06-22 14:59:35,057 [INFO] 인덱스 'fast-text-index'가 성공적으로 생성되었습니다!