Logstash를 활용하며 생긴 문제 해결과정
ELK
2025.09.16.
1. out of memory 발생
RDB -> elastic search에 적재를 하는데 데이터가 커서 그런지 Out of memory 발생!
Docker-compose.yml의 메모리의 크기를 증가시켰다.
services:
setup:
profiles:
- setup
build:
context: setup/
args:
ELASTIC_VERSION: ${ELASTIC_VERSION}
init: true
volumes:
- ./setup/entrypoint.sh:/entrypoint.sh:ro,Z
- ./setup/lib.sh:/lib.sh:ro,Z
- ./setup/roles:/roles:ro,Z
environment:
ELASTIC_PASSWORD: ${ELASTIC_PASSWORD:-}
LOGSTASH_INTERNAL_PASSWORD: ${LOGSTASH_INTERNAL_PASSWORD:-}
KIBANA_SYSTEM_PASSWORD: ${KIBANA_SYSTEM_PASSWORD:-}
METRICBEAT_INTERNAL_PASSWORD: ${METRICBEAT_INTERNAL_PASSWORD:-}
FILEBEAT_INTERNAL_PASSWORD: ${FILEBEAT_INTERNAL_PASSWORD:-}
HEARTBEAT_INTERNAL_PASSWORD: ${HEARTBEAT_INTERNAL_PASSWORD:-}
MONITORING_INTERNAL_PASSWORD: ${MONITORING_INTERNAL_PASSWORD:-}
BEATS_SYSTEM_PASSWORD: ${BEATS_SYSTEM_PASSWORD:-}
networks:
- elk
depends_on:
- elasticsearch
- kibana
- logstash
elasticsearch:
build:
context: elasticsearch/
args:
ELASTIC_VERSION: ${ELASTIC_VERSION}
volumes:
- ./elasticsearch/config/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml:ro,Z
- elasticsearch:/usr/share/elasticsearch/data:Z
ports:
- 9200:9200
- 9300:9300
environment:
node.name: elasticsearch
ES_JAVA_OPTS: -Xms8g -Xmx8g
build:
context: logstash/
args:
ELASTIC_VERSION: ${ELASTIC_VERSION}
volumes:
- ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml:ro,Z
- ./logstash/pipeline:/usr/share/logstash/pipeline:ro,Z
- ./logstash/config/pipelinse.yml:/usr/share/logstash/config/pipelines.yml:ro,Z
- ./logstash/input:/usr/share/logstash/input:ro,Z
- ./logstash/config/postgresql-42.7.2.jar:/usr/share/logstash/logstash-core/lib/jars/postgresql-42.7.2.jar
- ./logstash/state:/usr/share/logstash/state:rw,Z
ports:
- 5044:5044
- 50000:50000/tcp
- 50000:50000/udp
- 9600:9600
environment:
LS_JAVA_OPTS: -Xms20g -Xmx20g
LOGSTASH_INTERNAL_PASSWORD: ${LOGSTASH_INTERNAL_PASSWORD:-}
deploy:
resources:
limits:
memory: 24G
cpus: '8.0'
reservations:
memory: 16G
cpus: '4.0'
networks:
- elk
depends_on:
- elasticsearch
restart: unless-stopped
kibana:
build:
context: kibana/
args:
ELASTIC_VERSION: ${ELASTIC_VERSION}
volumes:
- ./kibana/config/kibana.yml:/usr/share/kibana/config/kibana.yml:ro,Z
ports:
- 5601:5601
environment:
KIBANA_SYSTEM_PASSWORD: ${KIBANA_SYSTEM_PASSWORD:-}
NODE_OPTIONS: --max-old-space-size=2048
networks:
- elk
depends_on:
- elasticsearch
restart: unless-stopped
networks:
elk:
driver: bridge
volumes:
elasticsearch:- 주의 사항으로는 logstash 만 메모리를 올리는게 아니라 Elastic search도 함께 메모리를 증가시켜줘야 한다.
- environment 섹션: 컨테이너 내부 환경변수 설정
- LS_JAVA_OPTS: -Xms20g -Xmx20g
- 무엇을 하는가?: Logstash는 Java로 만들어진 애플리케이션(JVM 위에서 동작)입니다. 이 설정은 Logstash가 사용하는 JVM의 힙(Heap) 메모리 크기를 지정
- -Xms20g: 최소 힙 메모리
- -Xmx20g: 최대 힙 메모리
- LS_JAVA_OPTS: -Xms20g -Xmx20g
- deploy 섹션: 컨테이너 외부 자원 할당 (Docker에게 지시)
- 이 컨테이너에 얼마만큼의 호스트 머신 자원(CPU, Memory)을 할당할지 지시합니다.
- Memory(16G ~ 24G)와 CPU(4.0 ~ 8.0) 할당
컨테이너 메모리 한도 (24GB) > JVM 최대 힙 크기 (20GB)
이렇게 힙 크기 + 여유 공간으로 설정해야, Logstash가 힙을 20GB까지 다 사용하더라도 컨테이너가 갑자기 종료되는 것을 막을 수 있다.
2. JDBC로 적재 시 너무 느린 속도
- 적재를 할 때 테이블에 count()함수를 실행 후, 쿼리문을 통해 데이터를 가져와 적재하는 방식
- view 테이블이라 한테이블은 적재가 불가능한 상황 (3700만건)
- 나머지 2테이블(1100만건, 600만건) → 소요시간 2시간
- 너무 느리니 csv로 시도
- python코드로 DB → csv : 가장 오래걸린 테이블 15분
- csv → logstash 적재 : 25분
- 약 40분
- 최적화 : 오래걸린다고 생각한 (postgrsql 공간함수)를 view 테이블에서 사용하지 말고 미리 계산해 넣어주자
- python코드로 DB → csv : 12분
- csv → logstash 적재 : 25분
- 약 35분
- DB → csv가 미세하게 빨라졌다
3. 테이블 동기화 중 pnu컬럼 뒤에 4자리 숫자 버림 처리
input {
file {
path => "/usr/share/logstash/input/elk_jibun_address_view_export.csv"
start_position => "beginning"
sincedb_path => "/usr/share/logstash/state/.sincedb_elk_jibun_address"
codec => "plain"
}
}
filter {
csv {
separator => ","
skip_header => true
columns => [
"jibun_address", "x", "y", "pnu"
]
}
mutate {
convert => { "x" => "float" }
convert => { "y" => "float" }
convert => { "pnu" => "string" }
strip => ["jibun_address"]
}
ruby {
code => "event.set('location', [event.get('x'), event.get('y')])"
}
}
output {
elasticsearch {
hosts => ["http://192.168.50.248:9200"]
user => "elastic"
password => "${LOGSTASH_INTERNAL_PASSWORD}"
index => "elk_jibun_address"
document_id => "%{pnu}"
manage_template => true
template_name => "elk_jibun_address_template"
template => "/usr/share/logstash/pipeline/jibun_address_template.json"
template_overwrite => true
}
# 진행 상황을 점(.)으로 확인
stdout {
codec => dots
}
}| 구분 | convert에 명시하지 않았을 때 (암시적 처리) | convert에 명시했을 때 (명시적 처리) |
|---|---|---|
| Logstash | pnu를 “숫자처럼 생긴 문자열”로 간주. JSON 생성 시 따옴표 없이 숫자처럼 보낼 위험이 있음. (일부 버전이나 환경에서) | pnu를 “반드시 문자열”로 확정. JSON 생성 시 따옴표를 붙여서 보냄. |
| Elasticsearch | 따옴표 없는 긴 숫자를 받고 long 타입으로 추측. 이 과정에서 정밀도 손실 발생 (뒷자리 0000). | 따옴표 있는 문자열을 받고 keyword 타입으로 정확히 인식. 데이터 손실 없음. |
| 결과 | 데이터 손상 발생 | 데이터 무결성 보장 |
4. 이름 검색 (N gram tokenizer)
“특정 단어들만 입력하더라도 검색이 되야한다.”
찾아보니 한국어 형태소 분석을 위한 nori가 있고, 토큰화 하는 n-gram, Edge n-gram이 있는데
일단 문자 단위로 토큰화(n-gram)하고 추후 개선하기로 하였다.
output을 작성할 때 elastic search에 template_name, template을 지정해주면 된다.
template에 어떻게 tokenizer할 지 설정한다.
input {
file {
path => "/usr/share/logstash/input/elk_local_data_view_export.csv"
start_position => "beginning"
sincedb_path => "/usr/share/logstash/state/.sincedb_elk_local_data"
codec => "plain"
}
}
filter {
csv {
separator => ","
skip_header => true
columns => [
"id",
"type",
"type_name",
"name",
"x",
"y",
"table_name",
"geom"
]
}
mutate {
convert => { "x" => "float" }
convert => { "y" => "float" }
strip => [
"id",
"type",
"type_name",
"name",
"table_name",
"geom"
]
}
if [x] and [y] {
ruby {
code => 'event.set("location", [event.get("x"), event.get("y")])'
}
}
}
output {
elasticsearch {
hosts => ["http://192.168.50.248:9200"]
user => "elastic"
password => "${LOGSTASH_INTERNAL_PASSWORD}"
index => "elk_local_data"
timeout => 60
document_id => "%{id}"
manage_template => true
template_name => "elk_local_data_template"
template => "/usr/share/logstash/pipeline/local_data_template.json"
template_overwrite => true
}
stdout {
codec => dots
}
file {
path => "/usr/share/logstash/logs/elk_local_data_debug.log"
codec => line { format => "ELK_DEBUG: %{@timestamp} - %{mgt_no}" }
}
}{
"index_patterns": ["elk_local_data*"],
"template": {
"settings": {
"index.max_ngram_diff": 10,
"analysis": {
"analyzer": {
"korean_ngram_analyzer": {
"tokenizer": "korean_ngram_tokenizer"
}
},
"tokenizer": {
"korean_ngram_tokenizer": {
"type": "ngram",
"min_gram": 2,
"max_gram": 10,
"token_chars": ["letter", "digit"]
}
}
}
},
"mappings": {
"properties": {
"id": { "type": "keyword" },
"type": { "type": "keyword" },
"type_name": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword"
}
}
},
"name": {
"type": "text",
"fields": {
"ngram": {
"type": "text",
"analyzer": "korean_ngram_analyzer"
},
"keyword": {
"type": "keyword"
}
}
},
"x": { "type": "float" },
"y": { "type": "float" },
"location": { "type": "geo_point" },
"table_name": { "type": "keyword" },
"geom": {
"type": "keyword",
"index": false
}
}
}
}
}5. 지도 화면 내 중심과 가까운 순으로 검색
이거 때문에 고생 좀 했다.
이 기능은 사용자가 지도를 움직일 때마다 보이는 영역 안에서, 입력한 키워드와 일치하면서 지도 중심에서 가까운 장소를 순서대로 보여줘야 했다.
고생 좀 한 이유는 텍스트 관련성 점수와 거리 점수라는 두가지 요소를 어떻게 자연스럽게 조합해 결과를 가져와야 했기 때문이다.
예시는 다음과 같다. 사용자가 지도 중심을 ‘강남역’ 근처에 두고 ‘스타벅스’를 검색했다고 가정하자.
다음과 같은 점수 계산 로직에 의해 강남역이 안떴다.
텍스트 검색 * 거리 점수 계산 이렇게 점수가 매겨지는데 이 곱하기가 문제였다.
- A 매장: 강남역에서 3km 떨어진, 화면에 보이는 스타벅스
- 텍스트 점수: 15.0
- 거리 점수: 0.01
- 최종 점수: 15.0 * 0.01 = 0.15점
- B 매장: 강남역에서 5km 떨어진, 화면에 보이는 스타벅스
- 텍스트 점수: 15.0
- 거리 점수: 0.0001
- 최종 점수: 15.0 * 0.0001 = 0.0015점
이런 느낌으로 이러한 현상을 ‘점수 소멸’ 현상이라고 한다.
이 문제를 해결하기 위해 buildAddressDslQuery 코드는 접근 방식을 완전히 바꿨다. 점수 섞지말고 각자 할일만 하게
- filter (선별 담당): “일단 지도 화면 안에 있는 애들만 가져온다.”
- geoBoundingBox 쿼리가 가장 먼저 동작
- 이 쿼리는 점수 계산 x -> 오직 “이 장소가 현재 보이는 지도 사각형 영역 안에 있는가, 없는가?”만 판단해서 결과 후보군을 먼저 걸러낸다.
- 이제 검색 대상은 사용자가 보고 있는 지도 화면 안으로 한정
- must (점수 계산 담당): “후보들 중에서 ‘스타벅스’를 찾고 점수를 매긴다.”
- filter를 통과한 ‘지도 화면 내의 장소들’ 중에서 multiMatchQuery가 ‘스타벅스’라는 키워드를 검색
- 이때 계산되는 점수는 오직 순수한 텍스트 관련성 점수 -> 거리 요소가 점수 계산에 전혀 개입하지 않으므로, ‘점수 소멸’ 현상이 원천적으로 발생하지 않는다.
private Query buildAddressDslQuery(List<String> fields, String keyword, BoundsDto bounds, Double centerLat, Double centerLon, int maxResults) {
// 원본 필드와 .ngram 필드를 모두 검색하도록 필드 목록을 동적으로 생성
List<String> searchFields = fields.stream()
.flatMap(field -> List.of(field + "^3", field + ".ngram").stream())
.collect(Collectors.toList());
NativeQueryBuilder nativeQueryBuilder = new NativeQueryBuilder();
nativeQueryBuilder.withMaxResults(maxResults);
nativeQueryBuilder.withQuery(q -> {
BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
boolQueryBuilder.must(m -> m
.multiMatch(mm -> mm
.query(keyword)
.fields(searchFields)
.operator(Operator.And)
)
);
if (bounds != null) {
boolQueryBuilder.filter(f -> f
.geoBoundingBox(gbb -> gbb
.field("location")
.boundingBox(bb -> bb
.tlbr(b -> b
.topLeft(g -> g.latlon(ll -> ll.lat(bounds.getMaxLat()).lon(bounds.getMinLon())))
.bottomRight(g -> g.latlon(ll -> ll.lat(bounds.getMinLat()).lon(bounds.getMaxLon())))
)
)
)
);
}
return q.bool(boolQueryBuilder.build());
});
if ((keyword == null || keyword.trim().isEmpty()) && centerLat != null && centerLon != null) {
SortOptions sortOptions = SortOptions.of(s -> s
.geoDistance(gds -> gds
.field("location")
.location(GeoLocation.of(gl -> gl.latlon(ll -> ll.lat(centerLat).lon(centerLon))))
.order(SortOrder.Asc)
)
);
nativeQueryBuilder.withSort(sortOptions);
}
return nativeQueryBuilder.build();
}먼저 범위를 좁히고(filter), 그 안에서 검색한다(must)는 방법으로 해결했다.
점수 가지고 AI랑 삽질하다가 잘 안되서 관점을 바꿔봤더니 해결 (한번에 좀 알려주지 ㅜ)
6. 특정 필드를 가진 데이터는 범위 검색 x 상위 노출, 그 외 필드들은 범위 검색 o
이전 문제들을 해결하고 데이터 적재 작업을 계속하는데 기능 요구사항이 자꾸 날라왔다…
심지어 까다로운 요구사항 ㅡㅡ
“사용자가 서울에서 ‘불국사’를 검색해도, 경주에 있는 불국사가 검색 결과에 나와야 합니다. 하지만 동네 빵집은 지금 보고 있는 지도 화면 안에서만 찾아야 하고요. 아, 그리고 불국사 같은 유명한 장소는 점수도 더 높게 쳐 상위에 노출되게 해주세요”
쿼리 핵심은 핵심은 “OR” 조건 bool 쿼리를 중첩해서 사용하면 문제 해결!
{
"size": 1000,
"query": {
"function_score": {
"query": {
"bool": {
"must": [
{
"multi_match": {
"query": "불국사",
"fields": ["name^3", "name.ngram"],
"operator": "AND"
}
}
],
"filter": [
{
"bool": {
"should": [
{
"terms": {
"type_name.keyword": [
"행정기관",
"사찰",
"대형유명유적",
"미술관",
"재래시장",
"산림욕장",
"계곡",
"박물관",
"빌라",
"아파트",
"초중고",
"대학교"
]
}
},
{
"bool": {
"must_not": [
{
"terms": {
"type_name.keyword": [
"행정기관",
"사찰",
"대형유명유적",
"미술관",
"재래시장",
"산림욕장",
"계곡",
"박물관",
"빌라",
"아파트",
"초중고",
"대학교"
]
}
}
],
"filter": [
{
"geo_bounding_box": {
"location": {
"top_left": {
"lat": 37.6,
"lon": 126.8
},
"bottom_right": {
"lat": 37.4,
"lon": 127.1
}
}
}
}
]
}
}
],
"minimum_should_match": "1"
}
}
]
}
},
"functions": [
{
"filter": {
"terms": {
"type_name.keyword": [
"행정기관",
"사찰",
"대형유명유적",
"미술관",
"재래시장",
"산림욕장",
"계곡",
"박물관",
"빌라",
"아파트",
"초중고",
"대학교"
]
}
},
"weight": 1.2
}
],
"boost_mode": "multiply"
}
}
}논리는 이렇다
검색 결과에 포함될 조건 = (A: 중요 장소 목록에 포함되거나) OR (B: 중요 장소 목록에 없으면서 지도 범위 안에 있거나)
minimum_should_match: “1”: 이게 바로 “OR” 스위치다. should 배열 안에 있는 조건들 중 하나만 맞아도 통과시켜 준다.
첫 번째 should: type_name이 내가 지정한 중요 목록(‘대학교’, ‘박물관’ 등)에 포함되는지 확인한다. 여기에 해당하면 바로 통과다.
두 번째 should: “목록에 속하지 않으면, 구역 안에 있니?” must_not으로 목록이 아니라는 걸 먼저 확인하고, geo_bounding_box 필터로 현재 지도 범위 안에 있는지 확인한다. 이 두 조건을 모두 만족해야 통과다.
이렇게 filter로 대상을 걸러낸 후에, function_score의 functions 파트가 목록을 찾아내내 weight: 1.2로 점수를 20% 향상시켜 준다.
private Query buildAddressDslQuery(List<String> fields, String keyword, BoundsDto bounds, Double centerLat, Double centerLon, int maxResults) {
// 원본 필드와 .ngram 필드를 모두 검색하도록 필드 목록을 동적으로 생성
List<String> searchFields = fields.stream()
.flatMap(field -> List.of(field + "^3", field + ".ngram").stream())
.collect(Collectors.toList());
// 점수를 높여주고, 범위 검색에서 제외할 type_name 목록
List<String> boostedTypes = List.of(
"행정기관", "사찰", "대형유명유적", "미술관", "재래시장",
"산림욕장", "계곡", "박물관", "빌라", "아파트", "초중고", "대학교"
);
NativeQueryBuilder nativeQueryBuilder = new NativeQueryBuilder();
nativeQueryBuilder.withMaxResults(maxResults);
// 헬퍼 메서드에 boostedTypes 리스트를 전달
BoolQuery coreQuery = createQueryForFunctionScore(searchFields, keyword, bounds, boostedTypes);
// List<String>을 List<FieldValue>로 변환 (function_score 용)
List<FieldValue> boostedTypeValues = boostedTypes.stream()
.map(FieldValue::of)
.collect(Collectors.toList());
// function_score 쿼리를 사용하여 점수 조작
nativeQueryBuilder.withQuery(q -> q
.functionScore(fs -> fs
.query(queryBuilder -> queryBuilder.bool(coreQuery))
// 점수 조작 함수 (이 로직은 그대로 유지)
.functions(f -> f
.filter(filterQuery -> filterQuery
.terms(t -> t
.field("type_name.keyword")
.terms(tq -> tq.value(boostedTypeValues))
)
)
.weight(1.2)
)
.boostMode(FunctionBoostMode.Multiply)
)
);
return nativeQueryBuilder.build();
}
// createQueryForFunctionScore 메서드
private BoolQuery createQueryForFunctionScore(List<String> searchFields, String keyword, BoundsDto bounds, List<String> boostedTypes) {
BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
// 1. must: 키워드 조건은 항상 필수
boolQueryBuilder.must(m -> m
.multiMatch(mm -> mm
.query(keyword)
.fields(searchFields)
.operator(Operator.And)
)
);
// 2. bounds가 null이면 필터링 로직을 적용하지 않는다.
if (bounds == null) {
return boolQueryBuilder.build();
}
// List<String>을 List<FieldValue>로 변환 (bool 쿼리용)
List<FieldValue> boostedTypeValues = boostedTypes.stream()
.map(FieldValue::of)
.collect(Collectors.toList());
// 3. filter: 조건부 범위 검색 로직을 적용
boolQueryBuilder.filter(f -> f
.bool(b -> b
// 아래 should 조건 중 하나만 만족하면 된다.
.should(s -> s // 조건 1: boostedTypes에 포함되는 경우 (범위 필터 없음)
.terms(t -> t
.field("type_name.keyword")
.terms(tq -> tq.value(boostedTypeValues))
)
)
.should(s -> s // 조건 2: boostedTypes에 포함되지 않으면서 AND 범위 내에 있는 경우
.bool(b2 -> b2
// 조건 2-1: boostedTypes에 포함되지 않아야 함
.mustNot(mn -> mn
.terms(t -> t
.field("type_name.keyword")
.terms(tq -> tq.value(boostedTypeValues))
)
)
// 조건 2-2: 지도 범위 내에 있어야 함
.filter(f2 -> f2
.geoBoundingBox(gbb -> gbb
.field("location")
.boundingBox(bb -> bb
.tlbr(b3 -> b3
.topLeft(g -> g.latlon(ll -> ll.lat(bounds.getMaxLat()).lon(bounds.getMinLon())))
.bottomRight(g -> g.latlon(ll -> ll.lat(bounds.getMinLat()).lon(bounds.getMaxLon())))
)
)
)
)
)
)
// should 절 중 최소 하나는 만족해야 함을 의미 (OR 조건)
.minimumShouldMatch("1")
)
);
return boolQueryBuilder.build();
}코드를 보면, 복잡한 필터링 로직을 createQueryForFunctionScore라는 헬퍼 메서드로 분리했다.
이 메서드가 바로 위 JSON에서 봤던 까다로운 filter 부분을 전담해서 만들어준다.
그리고 buildAddressDslQuery는 이 메서드가 만들어준 핵심 쿼리(coreQuery)를 받아서 function_score로 한번 더 감싸 점수를 조작하는 역할을 한다.
문제 해결 후기
갑작스럽게 우리 부서의 데이터를 기반으로 API를 만들어 외부에 제공한다는 임무가 떨어졌다.
거기다 해당 기능은 전담해 검색 퀄리티를 계속 향상시켜야 한다.
복잡하기로 악명 높은 Elasticsearch의 DSL Query는 Gemini에게 시키면 뚝딱 만들어줬다.
multi_match에 bool 쿼리를 섞고, function_score로 가중치를 주는 것까지, 그럴듯한 쿼리가 순식간에 완성됐다. 일단 기능은 돌아갔다.
결국 AI가 짜준 쿼리를 보며 “대충 잘 나오는 것 같으니 넘어가자”라고 넘어갔는데 잘 알지못하니 불안한 마음이 생겼다.
DSL Query 작성하는 부분은 AI도움을 받는다고 쳐도 어느정도 쿼리 튜닝이 되야하는데 쿼리가 너무 익숙치 않았다.
일단, 앞으로 구현할 기능들이 남아서 책도 한권사고 패스트캠퍼스 강의 구매까지 했다.
Elasticsearch 빨리 배우고, 제대로 업무에 적용해야겠다.
Reference
6.6.4 NGram, Edge NGram, Shingle - elk가이드북
N-gram tokenizer - elk doc
[Elasticsearch 적용기] #2 Elasticsearch 검색 정확도 높이기: Nori와 Edge n-gram을 활용한 검색 최적화 - chaewonni