Elasticsearch — 개념 & Python 구현 가이드

Overview

Elasticsearch란?

Elasticsearch(ES)는 Apache Lucene 기반의 오픈소스 분산 검색·분석 엔진입니다. JSON 문서를 저장하고, 역인덱스(inverted index) 구조를 이용해 전문 검색(Full-Text Search)을 밀리초 단위로 처리합니다. ELK 스택(Elasticsearch · Logstash · Kibana)의 핵심 엔진으로도 쓰입니다.

핵심 특징 요약: 실시간 검색 · 수평 확장(샤딩) · RESTful API · 스키마리스 JSON · 집계 분석(Aggregation)

역인덱스(Inverted Index)란?

일반 DB는 "문서 → 단어" 순으로 저장하지만, ES는 "단어 → 문서 목록"을 미리 만들어 둡니다. 덕분에 특정 단어를 포함한 문서를 O(1)에 가까운 속도로 찾습니다.

      개념
      역인덱스 구조
    

# 원본 문서
Doc 1: "Elasticsearch is fast"
Doc 2: "Elasticsearch is distributed"
Doc 3: "fast search engine"

# 역인덱스 (Lucene이 내부적으로 생성)
"elasticsearch" → [Doc1, Doc2]
"fast"           → [Doc1, Doc3]
"distributed"    → [Doc2]
"search"         → [Doc3]
    

Core Concepts

핵심 개념

🗂️

Index

문서의 논리적 집합. RDB의 테이블과 유사. 하나의 인덱스는 여러 샤드로 분산 저장됩니다.

📄

Document

JSON 형태의 데이터 단위. RDB의 행(row)에 해당. 각 문서는 고유한 _id를 가집니다.

🔀

Shard

인덱스를 나눈 물리적 단위. Primary Shard(쓰기)와 Replica Shard(읽기·복구)로 구분됩니다.

🖥️

Node & Cluster

Node는 단일 ES 인스턴스, Cluster는 Node의 집합. Master Node가 클러스터 상태를 관리합니다.

🗺️

Mapping

필드의 데이터 타입을 정의하는 스키마. text, keyword, integer, date, geo_point 등이 있습니다.

🔍

Analyzer

텍스트를 토큰으로 분리하는 파이프라인. Character Filter → Tokenizer → Token Filter 순으로 처리합니다.

Analyzer 파이프라인

      개념
      분석 흐름
    

# 입력 텍스트 → Analyzer 처리 순서

입력:  "Elasticsearch is AMAZING!"

1. Character Filter  → HTML 태그 제거, 특수문자 변환
2. Tokenizer        → ["Elasticsearch", "is", "AMAZING"]
3. Token Filter     → lowercase, stop words 제거
결과:  ["elasticsearch", "amazing"]  # 역인덱스에 저장
    

Comparison

RDB vs Elasticsearch

Elasticsearch의 용어는 관계형 DB와 대응 관계가 있습니다. 그러나 ES는 조인이 없고, 쓰기보다 읽기(검색) 최적화에 집중합니다.

RDBMS	Elasticsearch	설명
Database	Cluster	최상위 논리 단위
Table	Index	동일 유형 문서의 집합
Row	Document	단일 데이터 레코드 (JSON)
Column	Field	데이터의 속성
Schema	Mapping	필드 타입 정의
JOIN	nested / has_child	ES는 JOIN 지양, 비정규화 권장
Index (B-Tree)	Inverted Index	검색 방식의 근본적 차이
ACID 트랜잭션	단일 문서만 원자적	다중 문서 트랜잭션 없음

Setup

환경 설정

Elasticsearch 서버 실행 (Docker)

로컬 개발 환경에서 가장 빠른 방법

      bash
      Docker 실행
    

# 단일 노드 (개발용)
docker run -d \
  --name elasticsearch \
  -p 9200:9200 \
  -e "discovery.type=single-node" \
  -e "xpack.security.enabled=false" \
  -e "ES_JAVA_OPTS=-Xms512m -Xmx512m" \
  docker.elastic.co/elasticsearch/elasticsearch:8.13.0

# 실행 확인
curl http://localhost:9200
    

Python 라이브러리 설치

공식 클라이언트 설치 (버전을 ES 서버에 맞춰야 함)

      bash
      pip 설치
    

pip install elasticsearch==8.13.0
# 또는 최신 버전
pip install elasticsearch
    

Python 클라이언트 연결

기본 연결 및 인증 설정

      Python
      client_connect.py
    

from elasticsearch import Elasticsearch

# 기본 연결 (보안 비활성화 개발 환경)
es = Elasticsearch("http://localhost:9200")

# 인증이 있는 운영 환경
es = Elasticsearch(
    "https://my-es-host:9200",
    basic_auth=("elastic", "password"),
    ca_certs="/path/to/ca.crt",
)

# 연결 확인
print(es.info())
print(es.ping())  # True면 정상
    

CRUD Operations

인덱스 생성 및 CRUD

인덱스 생성 (Mapping 포함)

      Python
      create_index.py
    

# 상품 인덱스 생성
index_body = {
    "settings": {
        "number_of_shards": 3,
        "number_of_replicas": 1,
        "analysis": {
            "analyzer": {
                "korean_analyzer": {
                    "type": "custom",
                    "tokenizer": "nori_tokenizer",  # 한국어 형태소
                    "filter": ["lowercase"]
                }
            }
        }
    },
    "mappings": {
        "properties": {
            "name":        {"type": "text", "analyzer": "korean_analyzer"},
            "name_raw":    {"type": "keyword"},     # 정렬·집계용
            "price":       {"type": "integer"},
            "category":    {"type": "keyword"},
            "description": {"type": "text"},
            "created_at":  {"type": "date"},
            "rating":      {"type": "float"},
            "tags":        {"type": "keyword"}   # 배열도 동일 타입
        }
    }
}

if not es.indices.exists(index="products"):
    es.indices.create(index="products", body=index_body)
    print("인덱스 생성 완료")
    

문서 CRUD

      Python
      crud_operations.py
    

# ── Create (색인) ──────────────────────
doc = {
    "name": "에어팟 프로 2세대",
    "name_raw": "에어팟 프로 2세대",
    "price": 329000,
    "category": "이어폰",
    "description": "능동 소음 차단 기능이 향상된 무선 이어폰",
    "rating": 4.8,
    "tags": ["애플", "노이즈캔슬링", "무선"]
}

# ID 지정
resp = es.index(index="products", id="prod-001", document=doc)
print(resp["result"])  # "created"

# ID 자동 생성
resp = es.index(index="products", document=doc)
print(resp["_id"])  # 자동 생성된 UUID

# ── Read (조회) ──────────────────────
doc = es.get(index="products", id="prod-001")
print(doc["_source"])  # 실제 문서 내용

# ── Update (수정) ──────────────────────
# 일부 필드만 업데이트 (doc 방식)
es.update(
    index="products",
    id="prod-001",
    doc={"price": 299000, "rating": 4.9}
)

# 스크립트로 동적 업데이트
es.update(
    index="products",
    id="prod-001",
    script={
        "source": "ctx._source.price -= params.discount",
        "params": {"discount": 30000}
    }
)

# ── Delete (삭제) ──────────────────────
es.delete(index="products", id="prod-001")

# ── Bulk (대량 처리) ──────────────────
from elasticsearch.helpers import bulk

actions = [
    {"_index": "products", "_id": f"prod-{i}", "_source": {...}}
    for i in range(1000)
]
success, failed = bulk(es, actions)
print(f"성공: {success}, 실패: {len(failed)}")
    

Query DSL

쿼리 유형

ES는 JSON 기반의 Query DSL(Domain Specific Language)로 검색합니다. 크게 Full-Text Query(관련도 점수 계산)와 Term-level Query(정확한 일치)로 나뉩니다.

Full-Text match

분석된 텍스트 검색. 형태소 분석 후 매칭하므로 "노이즈캔슬링"으로 "노이즈 캔슬링"도 찾음.

Full-Text multi_match

여러 필드를 동시에 검색. 가중치(^) 부여로 중요도 조절 가능.

Term term / terms

keyword 필드 정확 일치. 대소문자 포함 그대로 비교. 주로 category, tag 필터에 사용.

Range range

숫자·날짜 범위 검색. gte, lte, gt, lt 연산자 지원.

Bool bool

must(AND), should(OR), must_not(NOT), filter 절을 조합해 복합 쿼리 구성.

Full-Text match_phrase

단어 순서까지 일치해야 함. "에어팟 프로"는 "에어팟"+"프로"가 인접해야 매칭.

match 쿼리

      Python
      match_query.py
    

# 기본 match 쿼리
resp = es.search(
    index="products",
    query={
        "match": {
            "description": {
                "query": "노이즈 캔슬링 무선",
                "operator": "or",  # or(기본) | and
                "fuzziness": "AUTO"  # 오타 허용
            }
        }
    },
    size=10,
    from_=0   # 페이지네이션
)

for hit in resp["hits"]["hits"]:
    print(f"{hit['_score']:.3f} | {hit['_source']['name']}")

# multi_match: 여러 필드 검색, 가중치 부여
resp = es.search(
    index="products",
    query={
        "multi_match": {
            "query": "에어팟",
            "fields": ["name^3", "description", "tags"]
            # name 필드 점수를 3배 가중치
        }
    }
)
    

Bool 복합 쿼리

      Python
      bool_query.py
    

# "이어폰" 카테고리에서 "노이즈캔슬링" 검색,
# 가격 100,000~500,000 사이, "애플" 태그 우선
resp = es.search(
    index="products",
    query={
        "bool": {
            "must": [
                {"match": {"description": "노이즈캔슬링"}}
            ],
            "filter": [  # 점수에 영향 없는 필터 (캐시됨 → 빠름)
                {"term":  {"category": "이어폰"}},
                {"range": {"price": {"gte": 100000, "lte": 500000}}}
            ],
            "should": [  # 매칭 시 점수 boost (optional)
                {"term": {"tags": "애플"}}
            ],
            "must_not": [
                {"term": {"tags": "단종"}}
            ]
        }
    },
    sort=[{"rating": {"order": "desc"}}]
)
    

Aggregation

집계 분석 (Aggregation)

Aggregation은 검색 결과를 통계적으로 분석합니다. Bucket(그룹화), Metric(평균·합계·최대), Pipeline(집계 결과 재집계) 세 종류가 있습니다.

      Python
      aggregation.py
    

# 카테고리별 상품 수 + 평균 가격 집계
resp = es.search(
    index="products",
    size=0,  # 검색 결과 문서는 필요 없음
    aggs={
        "by_category": {
            "terms": {"field": "category", "size": 10},
            "aggs": {  # 중첩 집계
                "avg_price": {"avg":   {"field": "price"}},
                "max_price": {"max":   {"field": "price"}},
                "price_hist": {
                    "histogram": {
                        "field":    "price",
                        "interval": 50000  # 5만원 단위 버킷
                    }
                }
            }
        },
        "rating_stats": {
            "stats": {"field": "rating"}
            # count, min, max, avg, sum 한번에
        }
    }
)

buckets = resp["aggregations"]["by_category"]["buckets"]
for b in buckets:
    print(f"{b['key']}: {b['doc_count']}개, 평균가 {b['avg_price']['value']:,.0f}원")
    

Advanced

고급 기능

Scroll / Search After — 대용량 페이지네이션

주의: from + size 방식은 10,000건 이상에서 성능이 급격히 저하됩니다. 대용량 데이터는 search_after를 사용하세요.

      Python
      search_after.py
    

# search_after: 커서 기반 페이지네이션
last_sort = None
all_docs = []

while True:
    body = {
        "size": 1000,
        "query": {"match_all": {}},
        "sort": [
            {"price": {"order": "asc"}},
            {"_id": {"order": "asc"}}   # tie-breaker
        ]
    }
    if last_sort:
        body["search_after"] = last_sort

    resp = es.search(index="products", **body)
    hits = resp["hits"]["hits"]
    if not hits:
        break
    all_docs.extend(hits)
    last_sort = hits[-1]["sort"]

print(f"총 {len(all_docs)}건 수집")
    

Highlight — 검색어 강조

      Python
      highlight.py
    

resp = es.search(
    index="products",
    query={"match": {"description": "노이즈 캔슬링"}},
    highlight={
        "fields": {
            "description": {
                "pre_tags":  ["<em>"],
                "post_tags": ["</em>"],
                "fragment_size": 150
            }
        }
    }
)

for hit in resp["hits"]["hits"]:
    highlight = hit.get("highlight", {}).get("description", [])
    print(highlight[0] if highlight else "")
# 예: "능동 <em>노이즈 캔슬링</em> 기능이 향상된 무선 이어폰"
    

Suggest — 자동완성 / 오타 교정

      Python
      suggest.py
    

# Completion Suggester (자동완성) 위한 매핑
es.indices.put_mapping(
    index="products",
    properties={
        "suggest": {"type": "completion"}
    }
)

# 자동완성 쿼리
resp = es.search(
    index="products",
    suggest={
        "name_suggest": {
            "prefix": "에어",
            "completion": {
                "field": "suggest",
                "size": 5
            }
        }
    }
)

for option in resp["suggest"]["name_suggest"][0]["options"]:
    print(option["text"])
    

Best Practices

운영 팁

⚡ text vs keyword

전문 검색엔 text, 정렬·집계·정확 매칭엔 keyword. 두 용도 모두 필요하면 multi-field(fields) 사용.

🔎 filter vs query

점수가 필요 없는 조건(category, 날짜 범위)은 반드시 filter에. 캐시되어 must보다 훨씬 빠름.

📦 Bulk API

단건 색인 루프 금지. 항상 bulk() 또는 parallel_bulk()로 배치 처리. 권장 배치 크기 500~1,000건.

🗺️ 샤드 수 설계

인덱스 생성 후 Primary Shard 수 변경 불가. 초기에 데이터 규모를 고려해 설정. 일반적으로 샤드당 10~50GB 권장.

🔄 Alias 활용

인덱스에 alias를 붙여 운영. 재색인(reindex) 시 alias를 원자적으로 전환해 무중단 업데이트 가능.

📊 모니터링

Kibana 또는 cat API로 샤드 상태, 힙 사용량, 검색 레이턴시 지속 모니터링. GC 압박이 지연의 주원인.

Alias를 이용한 무중단 재색인

      Python
      reindex_with_alias.py
    

# 1. 새 인덱스 생성 (v2)
es.indices.create(index="products_v2", body=new_mapping)

# 2. 데이터 복사
es.reindex(body={
    "source": {"index": "products_v1"},
    "dest":   {"index": "products_v2"}
}, wait_for_completion=False)  # 비동기

# 3. alias 원자적 전환 (무중단)
es.indices.update_aliases(actions=[
    {"remove": {"index": "products_v1", "alias": "products"}},
    {"add":    {"index": "products_v2", "alias": "products"}}
])

# 애플리케이션은 "products" alias만 사용 → 인덱스 변경 투명
    

한국어 검색 팁: 한국어 형태소 분석을 위해 nori 플러그인을 설치하세요. bin/elasticsearch-plugin install analysis-nori. nori_tokenizer와 nori_part_of_speech 필터 조합으로 조사·어미 제거가 가능합니다.

Elasticsearch 개념 & Python 구현

Elasticsearch란?

역인덱스(Inverted Index)란?

핵심 개념

Index

Document

Shard

Node & Cluster

Mapping

Analyzer

Analyzer 파이프라인

RDB vs Elasticsearch

환경 설정

Elasticsearch 서버 실행 (Docker)

Python 라이브러리 설치

Python 클라이언트 연결

인덱스 생성 및 CRUD

인덱스 생성 (Mapping 포함)

문서 CRUD

쿼리 유형

Full-Text match

Full-Text multi_match

Term term / terms

Range range

Bool bool

Full-Text match_phrase

match 쿼리

Bool 복합 쿼리

집계 분석 (Aggregation)

고급 기능

Scroll / Search After — 대용량 페이지네이션

Highlight — 검색어 강조

Suggest — 자동완성 / 오타 교정

운영 팁

⚡ text vs keyword

🔎 filter vs query

📦 Bulk API

🗺️ 샤드 수 설계

🔄 Alias 활용

📊 모니터링

Alias를 이용한 무중단 재색인