Transformers

Elasticsearch - Transformers 임베딩 입력 저장 (Local Mode)

개요

  • 엘라스틱서치에서 밀집 벡터 위한 매핑 구성 방법 살펴보기
  • 밀집 벡터가 저장될 embedding 이라는 필드 정의
  • Python 코드로 엘라스틱서치 코드 구현

코드 전체 시나리오

  • Elasticsearch에 연결 및 인덱스 초기화
  • dense_vector 타입으로 매핑 정의
  • 문서 배열 정의
  • BERT 모델을 이용해 각 문서를 벡터 임베딩
  • 벡터 포함 문서를 Elasticsearch에 색인

Elasticsearch 클라이언트 연결

  • 로컬에서 실행 중인 Elasticsearch 서버에 접속
  • basic_auth: 로그인 자격 (ID: elastic, PW: 123456)
  • verify_certs=False: 인증서 검증 생략 (로컬에서 SSL 없이 사용 시 편의용)
es_admin = Elasticsearch("http://localhost:9200", 
                         basic_auth=("elastic", "123456"), 
                         verify_certs=False)

Mapping 정의 및 인덱스 생성

  • dense_vector: 벡터 검색용 필드 (벡터 유사도 기반 검색 가능)
  • dims: BERT의 출력 벡터는 기본적으로 768차원이므로 그에 맞춤
mapping = {
    "properties": {
        "embedding": {
            "type": "dense_vector",
            "dims": 768  # BERT의 출력 벡터 차원 수
        }
    }
}

기존 인덱스 삭제 후 새로 생성

  • 기존에 있던 chapter-2 인덱스를 삭제 (중복 방지)
  • 새로운 인덱스를 위에서 정의한 벡터 매핑으로 생성
try:
    es_admin.indices.delete(index="chapter-2")
    print("기존 chapter-2 인덱스를 삭제했습니다.")
except:
    print("chapter-2 인덱스가 존재하지 않습니다.")

es_admin.indices.create(index="chapter-2", body={'mappings': mapping})
print("새로운 chapter-2 인덱스를 생성했습니다.")

색인할 문서 데이터 구성

  • titletext로 구성된 단순 문서 리스트
  • text는 BERT 임베딩의 입력값이 된다
docs = [
    {"title": "Document 1", "text": "This is the first document"},
    {"title": "Document 2", "text": "This is the second document"},
    {"title": "Document 3", "text": "This is the third document"}
]

BERT 모델과 토크나이저 초기화

  • bert-base-uncased: Hugging Face에서 사전 학습된 BERT 모델
  • AutoTokenizer: 입력 텍스트를 BERT가 이해할 수 있는 토큰으로 변환
  • AutoModel: 텍스트에 대한 BERT 임베딩 추출
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
model = AutoModel.from_pretrained("bert-base-uncased")

BERT 임베딩 생성

  • tokenizer(...): 텍스트를 토큰화하고 PyTorch 텐서로 변환
  • model(**inputs): BERT 실행 → 각 토큰에 대한 임베딩 벡터 반환
  • last_hidden_state.mean(dim=1): 문장의 전체 임베딩을 mean pooling으로 하나의 벡터로 압축 (1×768 벡터)
  • squeeze(0).numpy(): 불필요한 batch 차원 제거 후 NumPy로 변환
  • tolist(): Elasticsearch에 저장 가능하게 리스트 형태로 변환
for doc in docs:
    text = doc["text"]
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
    with torch.no_grad():
        outputs = model(**inputs).last_hidden_state.mean(dim=1).squeeze(0).numpy() 
        doc["embedding"] = outputs.tolist()

Elasticsearch에 색인

  • 각 문서를 chapter-2 인덱스에 색인
  • Elasticsearch는 embedding 필드를 dense_vector로 저장하며, 향후 벡터 검색에도 사용 가능
for doc in docs:
    es_admin.index(index="chapter-2", body=doc)

확인

  • Kibana | Management | Dev Tools에서 색인된 문서 조회
GET chapter-2/_search

image.png

엘라스틱 클라우드에 데이터 추가하기 - 예제 (2025, 06)

CH03 - 데이터 추가

개요

  • Cloud에 데이터 추가

이전 예제 확인

파이썬 코드

# 필요한 라이브러리들을 가져옵니다
import time                    # 시간 지연을 위한 라이브러리
import requests               # HTTP 요청을 위한 라이브러리
from bs4 import BeautifulSoup # HTML 파싱을 위한 라이브러리
from elasticsearch import Elasticsearch  # Elasticsearch 클라이언트

# ✅ Elastic Cloud 연결 (API 키 인증 방식)
# Elastic Cloud의 클러스터에 API 키를 사용하여 연결합니다
# API 키는 사용자명/비밀번호 대신 더 안전한 인증 방식입니다
es = Elasticsearch(
    "your_cloud_url",  # Elastic Cloud 클러스터 URL
    api_key="your_api_key"  # API 키
)

# 저장할 인덱스 이름을 상수로 정의합니다
INDEX_NAME = "evan-elk-search"

# ✅ 인덱스 생성 (존재하지 않으면 새로 생성)
# Elasticsearch에서 데이터를 저장할 인덱스가 있는지 확인하고, 없으면 새로 생성합니다
if not es.indices.exists(index=INDEX_NAME):
    es.indices.create(index=INDEX_NAME)  # 새 인덱스 생성
    print(f"✅ Index '{INDEX_NAME}' created.")
else:
    print(f"✅ Index '{INDEX_NAME}' already exists.")

# ✅ 명언 수집 함수 정의
def get_quotes():
    """
    quotes.toscrape.com 웹사이트에서 명언들을 수집하는 함수
    
    Returns:
        list: 수집된 명언 요소들의 리스트 (BeautifulSoup 객체들)
    """
    res = requests.get("http://quotes.toscrape.com")  # 웹사이트에 GET 요청
    soup = BeautifulSoup(res.text, "html.parser")     # HTML을 파싱
    return soup.select(".quote")                       # .quote 클래스를 가진 요소들을 선택하여 반환

# ✅ 30초 간격으로 명언들을 하나씩 저장
# 수집된 명언들을 가져옵니다
quotes = get_quotes()

# 각 명언을 순회하면서 Elastic Cloud에 저장합니다
for i, q in enumerate(quotes):
    # 명언 데이터를 딕셔너리 형태로 구성합니다
    doc = {
        "text": q.select_one(".text").text.strip(),      # 명언 텍스트 추출 (공백 제거)
        "author": q.select_one(".author").text.strip(),  # 저자 이름 추출 (공백 제거)
        "tags": [tag.text for tag in q.select(".tag")]   # 태그들을 리스트로 추출
    }
    
    # Elastic Cloud에 문서를 저장합니다
    res = es.index(index=INDEX_NAME, document=doc)
    print(f"[{i+1}] ✅ Saved to Elastic Cloud: {res['_id']}")  # 저장된 문서의 ID 출력
    
    # 30초 대기 (다음 명언 저장 전)
    time.sleep(30)
  • 클라우드에서 확인

image.png