Elasticsearch 대용량 데이터 파이프라인 구축기 - 수십억 건 데이터를 실시간으로 처리하는 법

 

문제 상황

어느 날 갑자기 팀원이 찾아와서 말했다. 

 

"지금 로그에서 사용자 선호 국가 데이터를 시각화하는데, 5개를 고른 유저와 1개만 고른 유저의 데이터가 같이 쌓이고 있어서 왜곡이 생기는 것 같아요. 국가별로 광고 캠페인을 집행해야하는데, 빨리 확인해볼 수 있을까요?"

 

맞는 말이었다. '한국'만을 선호하는 사용자와 5개 국가 중 하나로 한국을 포함시킨 사용자의 데이터가 동일한 가중치로 집계된다면, 실제 선호도를 제대로 반영하기 어려울 것이다.

 

"그럼, 선택한 개수에 따라 가중치를 계산해서 필드로 제공해드릴까요? 예를 들어 1개 선택하면 1.0, 5개 선택하면 각각 0.2씩 부여하는 식이면 정확하진 않아도, 지금보다는 훨씬 정확할 거에요." 

 

팀원도 좋은 아이디어라고 동의하며 바로 현재 상황을 분석하기 시작했다. 기존 Elasticsearch에는 이미 수십억 건의 로그가 쌓여있었고, user.preferences.preferRegion 필드는 존재했지만 가중치 필드는 없는 상황이었다. 더욱이 이 문제는 엄청난 데이터가 실시간으로 계속 유입되고 있다는 점이었다.

 

 

 

첫 번째 접근: Update By Query의 함정

처음 생각은 간단했다. "그냥 update_by_query로 기존 데이터를 업데이트하면 되겠네!"

POST /*-log-[service]-*/_update_by_query
{
  "script": {
    "source": """
    def preferRegion = ctx._source.user?.preferences?.preferRegion;
    def size = 0;
    
    if (preferRegion instanceof String) {
      size = (preferRegion == "ALL") ? 0 : 1;
    } else if (preferRegion instanceof List) {
      size = preferRegion.size();
    }
    
    if (size >= 5) ctx._source.preferRegion_weight = 0.2;
    else if (size == 4) ctx._source.preferRegion_weight = 0.25;
    // ... 이하 생략
    """
  }
}

테스트 인덱스로 돌려보니까... 1N억 건 처리해야하는데 대략 80시간이 걸릴 것 같다고 나왔다. 이거 운영에서 돌리면 서비스 터진다. 근데 로그 특성상 최신 로그를 반영하면 된다는 것을 알고, 최근 24시간 데이터부터 처리해볼까?" 싶어서 범위를 좁혀서 시도해봤다.

 

 

```json

POST /*-log-[service]-*/_update_by_query
{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-24h"
      }
    }
  },
  "script": {
    "source": "..."
  }
}

 

 

최근 하루 데이터는 대략 1000만 건 정도였는데, 이것도 30분은 걸렸다. 그리고 무엇보다 치명적인 문제가 있었다 - 새로 들어오는 데이터에는 여전히 가중치가 추가되지 않는다는 것! 매일 이런 식으로 배치를 돌려야 한다면 운영 부담이 너무 클 것 같았다.

 

 

 

두 번째 시도: Ingest Pipeline 발견

다른 방법을 찾아보다가 Ingest Pipeline이라는 걸 알게 되었다. 새로 들어오는 데이터에 자동으로 전처리를 적용하는 방식이었다.

 

json

PUT _ingest/pipeline/add_region_weight
{
  "description": "Add preferRegion_weight based on preferRegion size",
  "processors": [
    {
      "script": {
        "source": """
        if (ctx.user?.preferences?.preferRegion != null) {
          def preferRegion = ctx.user.preferences.preferRegion;
          def size = 0;
          
          if (preferRegion instanceof String) {
            size = (preferRegion == "ALL") ? 0 : 1;
          } else if (preferRegion instanceof List) {
            size = preferRegion.size();
          }
          
          if (size >= 5) ctx.preferRegion_weight = 0.2;
          else if (size == 4) ctx.preferRegion_weight = 0.25;
          else if (size == 3) ctx.preferRegion_weight = 0.33;
          else if (size == 2) ctx.preferRegion_weight = 0.5;
          else ctx.preferRegion_weight = 1.0;
        } else {
          ctx.preferRegion_weight = 1.0;
        }
        """
      }
    }
  ]
}

테스트해보니 잘 작동했다. 하지만 여기서 첫 번째 삽질이 시작되었다.

 

 

 

세 번째 난관: Index Template 우선순위 지옥

파이프라인만 만들어서는 새로 생성되는 인덱스에 자동으로 적용되지 않았다. Index Template이 필요했다.

 

json

PUT _index_template/log-service-template
{
  "index_patterns": ["*-log-[service]-*"],
  "priority": 200,
  "template": {
    "settings": {
      "index.default_pipeline": "add_region_weight"
    }
  }
}

그런데 이게 웬걸? 에러가 터졌다.

 

index template [log-service-template] has index patterns [*-log-[service]-*] 
matching patterns from existing templates with patterns that have the same priority [200]

알고 보니 Elasticsearch에는 수많은 기본 템플릿들이 있었고, 다들 200을 사용하고 있었다. 500으로 올려도 안 되고, 1000으로 올려도 안 되고... 결국 3000으로 올려서야 성공했다.

 

json

{
  "index_patterns": ["*-log-[service]-*"],
  "priority": 3000,
  "template": {
    "settings": {
      "index.default_pipeline": "add_region_weight"
    }
  }
}

 

 

 

네 번째 깨달음: 기존 인덱스는 따로 처리해야 한다

Template은 새로 생성되는 인덱스에만 적용된다는 걸 뒤늦게 알았다. 기존 인덱스들은 수동으로 파이프라인을 적용해야 했다.

 

json

PUT /*-log-[service]-*/_settings
{
  "index.default_pipeline": "add_region_weight"
}

다행히 이 작업은 메타데이터만 수정하는 거라 몇 초 만에 끝났다.

가장 중요한 건 제대로 작동하는지 확인하는 것이었다. _simulate API를 사용해서 파이프라인을 테스트해봤다.

 

json

POST _ingest/pipeline/add_region_weight/_simulate
{
  "docs": [
    {
      "_source": {
        "user": {
          "preferences": {
            "preferRegion": ["KR", "US"]
          }
        }
      }
    }
  ]
}

결과를 보니 정확히 0.5가 나왔다. 드디어 제대로 작동한다!

 

 

 

운영 모니터링과 지속적인 관리

파이프라인을 적용한 후에는 지속적인 모니터링이 필요했다. 실제 데이터에 필드가 잘 추가되고 있는지 확인하는 쿼리를 만들었다.

 

json

GET /*-log-[service]-*/_search
{
  "query": {
    "range": {
      "@timestamp": {
        "gte": "now-1h"
      }
    }
  },
  "_source": ["@timestamp", "preferRegion_weight", "user.preferences.preferRegion"],
  "size": 10
}

가중치 분포도 확인해서 의도한 대로 처리되고 있는지 체크했다.

 

json

GET /*-log-[service]-*/_search
{
  "size": 0,
  "aggs": {
    "weight_distribution": {
      "terms": {
        "field": "preferRegion_weight",
        "size": 10
      }
    }
  }
}

 

 

 

마지막 함정: Kibana에서 필드가 안 보인다

Kibana Discover에서 새 필드가 보이지 않았다. 알고 보니 Data View에서 필드 목록을 새로고침해야 했다. Stack Management → Data Views → Refresh field list를 누르니까 드디어 필드가 보였다.

 

 

 

마치며

이번 작업을 통해 배운 것들은 아래와 같다.

 

Ingest Pipeline 잘쓰자: Update By Query로 몇 시간 걸릴 작업을 실시간으로 처리할 수 있다. 새로 들어오는 데이터에 자동으로 적용되니까 한 번 설정해놓으면 끝이다.

Index Template 우선순위: Elasticsearch 기본 템플릿들이 생각보다 많다. 사용자 정의 템플릿은 1000 이상의 높은 우선순위를 사용하는 게 안전하다.

단계별 검증의 중요성: _simulate API로 파이프라인 로직을 미리 검증하고, 실제 데이터로 테스트하고, Kibana UI까지 확인하는 전체 플로우를 검증해야 한다.

지속적 모니터링: 파이프라인을 적용한 후에도 실제 데이터 적용률, 성능 영향, 가중치 분포 등을 지속적으로 모니터링해야 한다.

결국 처음 예상했던 것보다 훨씬 복잡했지만, 한 번 제대로 구축해놓으니 매일 수백만 건의 데이터가 자동으로 전처리되고 있다.