[Shoot-Pointer] 게시물 검색 성능 개선 - ElasticSearch 도입기(1) (Docker를 이용한 ElasticSearch + Kibana 구축 및 테스트 환경 구축)

    /**
     * 1. 제목 + 내용 게시물 조회 - NoOffset+slice 방식
     * 2. 조회된 게시물 최신순 정렬 반환.
     */
    @Query(value = """
        SELECT *
        FROM post p
        WHERE (p.title LIKE CONCAT('%', :search, '%')
            OR p.content LIKE CONCAT('%', :search, '%'))
          AND p.post_id < :lastPostId
        ORDER BY p.post_id DESC
        LIMIT :size
        """,
            nativeQuery = true)
    List<PostEntity> getPostEntitiesByPostTitleOrPostContentOrderByCreatedAtDesc(
            @Param("search") String search,
            @Param("size") int size,
            @Param("lastPostId") Long postId
    );

services:
  # Spring Boot Application
  shootpointer:
    build:
    depends_on:
      elasticsearch:
        condition: service_healthy
    environment:
      SPRING_ELASTICSEARCH_URIS: http://elasticsearch:9200
    ports:
      - "9000:9000"
    networks:
      - spring-network

  ## ElasticSearch
  elasticsearch:
    build:
      context: .
      dockerfile: Dockerfile.elesticsearch
    image: docker.elastic.co/elasticsearch/elasticsearch:8.6.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - logger.level=debug
      - ES_JAVA_OPTS=-Xms2g -Xmx2g
    ports:
      - "9200:9200"
      - "9300:9300"
    volumes:
      - esdata:/usr/share/elasticsearch/data
      - ./es-logs:/usr/share/elasticsearch/logs
    networks:
      - spring-network
    restart: always
    healthcheck:
      test: [ "CMD", "curl", "-f", "http://localhost:9200/_cluster/health" ]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 60s

  kibana:
      image: docker.elastic.co/kibana/kibana:8.6.0
      container_name: kibana
      environment:
        - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
      ports:
        - "5601:5601"
      depends_on:
        - elasticsearch
      networks:
        - spring-network
      restart: always

networks:
  spring-network:
    driver: bridge

volumes:
  esdata:

curl -XPOST "http://localhost:9200/_bulk" \
  -H "Content-Type: application/json" \
  --data-binary @/Users/kimdoyeon/bulk.json

curl: option --data-binary: out of memory
curl: try 'curl --help' or 'curl --manual' for more information

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

# 기존 nori 플러그인이 있으면 삭제 후 설치
RUN if [ -d "/usr/share/elasticsearch/plugins/analysis-nori" ]; then \
      bin/elasticsearch-plugin remove analysis-nori; \
    fi && \
    bin/elasticsearch-plugin install analysis-nori --batch

//ElasticSearch
implementation 'org.springframework.boot:spring-boot-starter-data-elasticsearch'

spring:
  # ElasticSearch
  elasticsearch:
    enabled: false
    uris: ${SPRING_ELASTICSEARCH_URIS:http://elasticsearch:9200}

@Configuration
@EnableElasticsearchRepositories
@Profile("dev")  // dev 프로파일에서만 활성화
public class ElasticSearchConfig  {
    @Value("${spring.elasticsearch.uris}")
    private String host;

    @Bean
    public ClientConfiguration clientConfiguration() {
        return ClientConfiguration.builder()
                .connectedTo(host.replace("http://",""))
                .build();
    }

    @Bean
    public ElasticsearchClient elasticsearchClient(){
        RestClient restClient=RestClient.builder(HttpHost.create(host)).build();
        return new ElasticsearchClient(new RestClientTransport(restClient,new JacksonJsonpMapper()));
    }
}

package com.midas.shootpointer.domain.post.entity;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.AccessLevel;
import lombok.Builder;
import lombok.Getter;
import lombok.NoArgsConstructor;
import org.springframework.context.annotation.Profile;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.*;

import java.time.LocalDateTime;
@Profile("!dev")  // dev 프로파일이 아닐 때만 활성화
@Getter
@Document(indexName = "post",createIndex = true)
@Mapping(mappingPath = "elasticsearch/post-mapping.json")
@Setting(settingPath = "elasticsearch/post-setting.json")
@JsonIgnoreProperties(ignoreUnknown = true)
@NoArgsConstructor(access = AccessLevel.PROTECTED)
public class PostDocument {
    @Id
    @Field(type = FieldType.Long)
    private Long postId;

    @Field(type = FieldType.Text)
    private String title;

    @Field(type = FieldType.Text)
    private String content;

    @Field(type = FieldType.Keyword)
    private HashTag hashTag;

    @Field(type = FieldType.Long)
    private Long likeCnt;

    @Field(type = FieldType.Text)
    private String memberName;

    @Field(type = FieldType.Date,format = {DateFormat.date_hour_minute_second_millis, DateFormat.epoch_millis})
    private LocalDateTime createdAt;

    @Field(type = FieldType.Date,format = {DateFormat.date_hour_minute_second_millis, DateFormat.epoch_millis})
    private LocalDateTime modifiedAt;

    @Field(type = FieldType.Text)
    private String highlightUrl;

    @Builder
    public PostDocument(Long postId,
                        String title,
                        String content,
                        HashTag hashTag,
                        Long likeCnt,
                        String memberName,
                        LocalDateTime createdAt,
                        LocalDateTime modifiedAt,
                        String highlightUrl
    ){
        this.content=content;
        this.hashTag=hashTag;
        this.postId=postId;
        this.title=title;
        this.memberName=memberName;
        this.likeCnt=likeCnt;
        this.createdAt=createdAt;
        this.modifiedAt=modifiedAt;
        this.highlightUrl=highlightUrl;
    }
}

public interface PostElasticSearchRepository extends ElasticsearchRepository<PostDocument,Long> {
}

@Component
@Profile("!dev")  // dev 프로파일이 아닐 때만 활성화
public class PostElasticSearchMapperImpl implements PostElasticSearchMapper {
    @Override
    public PostResponse docToResponse(PostDocument doc) {
        return PostResponse.builder()
                .title(doc.getTitle())
                .content(doc.getContent())
                .createdAt(doc.getCreatedAt())
                .memberName(doc.getMemberName())
                .hashTag(doc.getHashTag())
                .highlightUrl(doc.getHighlightUrl())
                .likeCnt(doc.getLikeCnt())
                .modifiedAt(doc.getModifiedAt())
                .postId(doc.getPostId())
                .build();
    }

    @Override
    public PostDocument entityToDoc(PostEntity post) {
        return PostDocument.builder()
                .content(post.getContent())
                .title(post.getTitle())
                .hashTag(post.getHashTag())
                .postId(post.getPostId())
                .likeCnt(post.getLikeCnt())
                .memberName(post.getMember().getUsername())
                .createdAt(post.getCreatedAt())
                .modifiedAt(post.getModifiedAt())
                .highlightUrl(post.getHighlight().getHighlightURL())
                .build();
    }
}

@Component
@Profile("!dev")  // dev 프로파일이 아닐 때만 활성화
@RequiredArgsConstructor
public class PostElasticSearchHelperImpl implements PostElasticSearchHelper{
    private final PostElasticSearchUtil postElasticSearchUtil;

    @Override
    public Long createPostDocument(PostEntity post) {
        return postElasticSearchUtil.createPostDocument(post);
    }
}

@RequiredArgsConstructor
@Component
@Profile("!dev")  // dev 프로파일이 아닐 때만 활성화
public class PostElasticSearchUtilImpl implements PostElasticSearchUtil{
    private final PostElasticSearchRepository postElasticSearchRepository;
    private final PostElasticSearchMapper mapper;

    @Transactional
    @Override
    public Long createPostDocument(PostEntity post) {
        PostDocument postDocument=mapper.entityToDoc(post);
        return postElasticSearchRepository.save(postDocument).getPostId();
    }
}

@Component
@Profile("testdata")
@RequiredArgsConstructor
public class DummyDataLoader implements CommandLineRunner {
    private final JdbcTemplate jdbcTemplate;
    private final MakeRandomWord makeRandomWord;
    private final MemberCommandRepository memberRepository;
    private final HighlightCommandRepository highlightCommandRepository;
    private final PostElasticSearchRepository postElasticSearchRepository;
    private final PostElasticSearchMapper mapper;

    private final int batchSize=10_000;
    private final int insertSize=100_000;
    private final PostQueryRepository postQueryRepository;

    @Override
    public void run(String... args) throws Exception {
        ...DB batch 로직

        /**
         * Elastic Search 배치 처리
         */
        long esStart = System.currentTimeMillis();
        System.out.println("ES 배치 시작 시간: " + LocalDateTime.now());
        int page = 0;

        Page<PostEntity> result;
        do {
            result  = postQueryRepository.findAllWithMemberAndHighlight(PageRequest.of(page, batchSize));

            // PostEntity → PostDocument 변환
            List<PostDocument> docs = result.stream()
                    .map(mapper::entityToDoc)
                    .toList();

            postElasticSearchRepository.saveAll(docs);

            System.out.println("ES 배치 : " + ((page + 1) * batchSize) + "건 삽입 완료");
            page++;
        } while (result.hasNext());

        long esEnd = System.currentTimeMillis();
        System.out.println("ES 배치 종료 시간: " + LocalDateTime.now());
        System.out.println("ES 전체 소요 시간(ms): " + (esEnd - esStart));
    }
}

import psycopg2
import json
import requests
import re
import time
import os

DB_CONFIG = {
    "dbname": os.getenv("DB_NAME", ""),
    "user": os.getenv("DB_USER", ""),
    "password": os.getenv("DB_PASS", ""),
    "host": os.getenv("DB_HOST", "postgres"),
    "port": int(os.getenv("DB_PORT", 5432)),
}
ES_URL = os.getenv("ES_URL", "http://elasticsearch:9200/posts/_bulk")
BATCH_SIZE = int(os.getenv("BATCH_SIZE", 1000))  

CTRL_RE = re.compile(r"[\x00-\x1F\x7F\u00A0]")

def clean_text(text: str) -> str:
    if not text:
        return None
    
    return CTRL_RE.sub(" ", text).strip() or None

def fetch_posts():
    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()
    cur.execute("""
        SELECT post_id, title, content, like_cnt, member_id, highlight_id,
               hash_tag, created_at, modified_at
        FROM post
    """)
    while True:
        rows = cur.fetchmany(BATCH_SIZE)
        if not rows:
            break
        yield rows
    cur.close()
    conn.close()

def bulk_upload(rows):
    bulk_lines = []
    for row in rows:
        post_id, title, content, like_cnt, member_id, highlight_id, hash_tag, created_at, modified_at = row

        meta = {"index": {"_id": str(post_id)}}
        doc = {
            "postId": post_id,
            "title": clean_text(title),
            "content": clean_text(content),
            "likeCnt": like_cnt,
            "memberId": str(member_id) if member_id else None,
            "highlightId": str(highlight_id) if highlight_id else None,
            "hashTag": hash_tag,
            "createdAt": created_at.isoformat() if created_at else None,
            "modifiedAt": modified_at.isoformat() if modified_at else None,
        }

        bulk_lines.append(json.dumps(meta, ensure_ascii=False))
        bulk_lines.append(json.dumps(doc, ensure_ascii=False))

    data = "\n".join(bulk_lines) + "\n"
    response = requests.post(ES_URL, headers={"Content-Type": "application/x-ndjson"}, data=data.encode("utf-8"))
    return response.json()

def main():
    start_time = time.time()  
    
    total, success, fail = 0, 0, 0
    for rows in fetch_posts():
        total += len(rows)
        resp = bulk_upload(rows)
        if not resp.get("errors"):
            success += len(rows)
            print(f"성공 - {len(rows)} docs (total {success}/{total})")
        else:
            fail += len(rows)
            print(f"실패 - {len(rows)} docs (total {fail}/{total})")
            print(json.dumps(resp, indent=2, ensure_ascii=False)[:1000])  # 일부만 출력
    
    end_time = time.time()  
    elapsed_ms = int((end_time - start_time) * 1000)

    print("=== 결과 요약 ===")
    print(f"총 문서: {total}")
    print(f"성공:    {success}")
    print(f"실패:    {fail}")
    print(f"총 소요 시간: {elapsed_ms} ms")

if __name__ == "__main__":
    main()

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY es_bulk.py .

CMD ["python", "es_bulk.py"]

psycopg2-binary
requests

docker build -t tkv00/es-bulk:latest .
docker login
docker push tkv00/es-bulk:latest

#!/bin/bash
docker pull tkv00/es-bulk:latest
docker run --rm \
  --network=shootpointer_be_spring-network \
  -e DB_NAME=shootpointer \
  -e DB_USER=myuser \
  -e DB_PASS=rlaehdus00 \
  -e DB_HOST=postgres \
  -e DB_PORT=5432 \
  -e ES_URL=http://elasticsearch:9200/post/_bulk \
  tkv00/es-bulk:latest