엘라스틱서치에서 인덱스의 데이터 집계와 통계를 추출하고싶다.

이를 위해서 찾아본결과 엘라스틱서치에서는 Transforming data라는 기능을 제공하고있다.

공식 문서는 아래
https://www.elastic.co/guide/en/elasticsearch/reference/8.17/transforms.html

개념

1. 데이터 집계: 특정 필드 값을 기준으로 데이터를 그룹화하고 통계를 계산할 수 있다
   ex) 날짜별로 뉴스 기사를 집계하거나, 특정 카테고리의 매출을 계산
2. 결과 저장: Transform은 결과 데이터를 새로운 인덱스에 저장하여, 별도로 분석하거나 검색할 수 있게 한다
3. 증분 처리: Transform은 새로 들어온 데이터만 증분 방식으로 처리할 수 있어 효율적이다
4. 주기적 실행: Transform은 설정된 주기(예: 1분, 1시간, 1일)에 따라 자동으로 실행되도록 설정할 수 있다

구조

Transform의 설정 쿼리는 크게 아래와 같은 구조로 이루어진다.
1. Source (소스 인덱스): 데이터를 가져올 인덱스나 쿼리
2. Dest (결과 인덱스): Transform 결과를 저장할 인덱스
3. Pivot (집계 기준): 데이터를 그룹화하거나 통계 계산을 수행할 기준과 방법
4. Sync (증분 설정): Transform이 새 데이터만 처리하도록 설정

이를 json형식으로 나타내면 아래와 같다

PUT _transform/example_transform
{
  "source": {
    "index": "example_source_index",
    "query": {
      "range": {
        "creation_time": {
          "gte": "2024-01-01T00:00:00",
          "lte": "2024-12-31T23:59:59"
        }
      }
    }
  },
  "dest": {
    "index": "example_result_index"
  },
  "frequency": "1h",
  "pivot": {
    "group_by": {
      "date": {
        "date_histogram": {
          "field": "creation_time",
          "calendar_interval": "day"
        }
      },
      "category": {
        "terms": {
          "field": "category.keyword"
        }
      }
    },
    "aggregations": {
      "total_count": {
        "value_count": {
          "field": "id"
        }
      },
      "top_keywords": {
        "terms": {
          "field": "keywords.keyword",
          "size": 5
        }
      }
    }
  },
  "sync": {
    "time": {
      "field": "creation_time",
      "delay": "30s"
    }
  }
}

Source

Transform의 소스 데이터는 source.index에 지정된 인덱스에서 가져온다. 이때 데이터를 필터링하거나 특정 조건에 맞는 데이터만 처리하고 싶다면 query를 추가할 수 있다

기존 엘라스틱서치의 _search와 유사한 기능이다.

"source": {
    "index": "example_source_index",
    "query": {
      "range": {
        "creation_time": {
          "gte": "2024-01-01T00:00:00",
          "lte": "2024-12-31T23:59:59"
        }
      }
    }
  }

• index
  - 데이터를 가져오는 인덱스, 만약 여러 인덱스를 가져오고 싶다면 아래와 같이 작성 가능하다.
  "index": ["index1", "index2", "log-*"]
• query
  - _search의 query와 동일한 구조이다.
  • 필터링을 이용해 효율을 높일 수 있다
• runtime_mappings
  - _search의 runtime_mappings와 동일하게 동작
  - Transform중 계산된 필드를 정의해서 원본 데이터에 추가적인 계산값을 포함할 수 있다.
  

Dest

Transform의 결과를 저장하는 인덱스의 이름이다

"dest": {
    "index": "example_result_index"
  }

Frequency

어느 주기로 데이터를 수집하여 통계를 낼지 정한다.

"frequency": "1h",
// 1s, 1m, 1h, 1d

Pivot

데이터를 그룹화하고 집계를 수행하는 기준을 생성한다.

"pivot": {
    "group_by": {
      "date": {
        "date_histogram": {
          "field": "creation_time",
          "calendar_interval": "day"
        }
      },
      "category": {
        "terms": {
          "field": "category.keyword"
        }
      }
    },
    "aggregations": {
      "total_count": {
        "value_count": {
          "field": "id"
        }
      },
      "top_keywords": {
        "terms": {
          "field": "keywords.keyword",
          "size": 5
        }
      }
    }
  },

• group_by : SQL에서 GROUP_BY와 동일하게 데이터를 특정 그룹으로 묶어 통계를 산출한다.
• aggregations : 그룹화된 데이터에서 특정 필드에 대한 통계를 계산한다

Sync

Transform이 새로 들어온 데이터만을 처리하기 위해 존재하는 설정 항목이다.

"sync": {
    "time": {
      "field": "creation_time",
      "delay": "30s"
    }
  }

현재 설정에 따르면 creation_time을 기준으로 통계를 수집중이다.
해당 creation_time을 기준으로 통계된 문서의 최대 시간을 기록한다.
그리고 다음 통계부터는 해당 값 이후의 데이터부터 처리한다.

• field : 시간을 수집하기위한 기준 필드
• delay : 지연 시간, 엘라스틱서치에서 데이터를 인덱싱하는 시간이 지연될 수 있기때문에 해당 지연 문제를 해결하고 안정화가 된 다음 수집을 하기위해 설정한다.

과거의 데이터 업데이트

결론부터 말하면 불가능하다

Sync설정이 존재하고 time.field의 기준 필드 최대시간을 기준으로 탐색하여 그 다음 시간부터 수집을 하기때문에 임의로 과거의 값을 넣어도 데이터로 통계가 되지 않는다.

Transform 관련 요청 및 수정, 삭제

기본적으로 이미 돌아가고있는 Transform을 수정할 수는 없기때문에
Transform을 멈춘다음, 삭제후 설정을 수정하고 재생성해야한다.

• Transform 시작

POST _transform/{transform_id}/_start

• Transform 정지

POST _transform/{transform_id}/_stop

• Transform 삭제

DELETE _transform/{transform_id}

• Transform 상태 확인

GET _transform/{transform_id}/_stats

결론

Transforming data는 데이터를 효율적으로 수집 집계하여 통계 분석을 할 수 있다.
하지만 과거의 데이터가 들어가면 수집을 하지 못하기때문에
과거에 데이터가 추가될 일이없는 인덱스에서 편하게 통계를 내고싶으면 활용할 수 있다.