Elasticsearch data stream

Data streams

Data Stream은 시계열 데이터를 관리하기 위한 기능으로 하나 이상의 시계열 인덱스를 논리적으로 묶은 그룹으로, 여러 시계열 데이터 소스를 하나의 개체로 인지하여 관리하기 때문에 로그, 이벤트, 지표 및 기타 지속적으로 생성되는 데이터에 적합한 기능입니다. Elasticsearch의 경우 역색인 형태를 가지고 있기 때문에 Document를 삭제하거나 업데이트를 하는데 많이 비율적인 모습을 보이고 있습니다. 그렇기 때문에 데이터를 Document 단위로 관리를 하는 것보다는 인덱스 단위로 관리하는 것이 더 효율적입니다. 이번 문서에서는 시계열 데이터를 data stream을 통해 관리하는 방법에 대하여 정리하였습니다.

 

Data stream의 인덱싱과 검색 동작 방식

data stream을 통해 데이터를 인덱싱 할 경우 마지막으로 생성된 인덱스에 데이터를 인덱싱을 수행합니다.

그러나 검색은 인덱싱과 다르게 data stream으로 생성된 모든 인덱스에 검색 요청을 수행합니다.

사전 작업(클러스터 구성 및 설정)

먼저 이번 실습을 위해 저는 Master Node 3대, Hot Data 2대, Warm Data 2대로 클러스터를 구성하였으며, 빠른 테스트를 위해 클러스터가 인덱스를 롤오버 검사를 하는 주기를 5초로 설정하였습니다. 설정 방법은 아래와 같습니다.

### 인덱스 롤오버 체크 주기 변경
PUT http://localhost:9200/_cluster/settings
Content-Type: application/json

{
  "persistent": {
    "indices.lifecycle.poll_interval": "5s"
  }
}

 

ILM(Index Lifecycle Management) 설정

data stream은 ILM(Index Lifecycle Management)를 고려하여 추가된 기능으로 data stream을 사용할 때 ILM 기능과 같이 사용하는 것을 권장합니다. ILM 설정으로는 빠른 테스트를 위해 인덱스가 롤오버 되면 2분 뒤에 Warm 노드로 이동하고 10분 뒤에 Delete 되도록 설정하였습니다. 실제로는 이렇게 짧은 시간으로 설정하지는 않고 최고 일단위로 설정합니다.

### Create ILM policy
PUT http://localhost:9200/_ilm/policy/my-ilm-policy
Content-Type: application/json

{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "1gb",
            "max_docs": 10
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "2m",
        "actions": {
          "forcemerge": {
            "max_num_segments": 1
          },
          "shrink": {
            "number_of_shards": 1
          },
          "allocate": {
            "number_of_replicas": 1
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "delete": {
        "min_age": "10m",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

데이터가 Warm노드로 이동할 때 shrink 옵션을 통하여 샤드의 개수가 1개로 설정되도록 하였습니다.

 

Component template 생성

component template은 template의 기능을 여러개로 분리하여 관리하는 기능으로 한번 정의해 놓으면 다른 인덱스 template에서도 재사용할 수 있게 하는 기능입니다. 이번 실습에서는 settings에 관련된 template과 mapping에 관련된 template을 분리하여 생성하였습니다.

데이터 스트림에 인덱싱된 모든 문서에는 또는 필드 유형 @timestamp으로 매핑된 필드가 포함되어야 합니다. 인덱스 템플릿이 필드에 대한 매핑을 지정하지 않으면 Elasticsearch는 기본 옵션을 사용하여 필드로 매핑합니다

### create component template
PUT http://localhost:9200/_component_template/my-settings
Content-Type: application/json

{
  "template": {
    "settings": {
      "index.lifecycle.name": "my-ilm-policy",
      "number_of_shards": "2",
      "number_of_replicas": "1",
      "refresh_interval": "5s"
    }
  }
}

### create component template
PUT http://localhost:9200/_component_template/my-mappings
Content-Type: application/json

{
  "template": {
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date",
          "format": "date_optional_time||epoch_millis"
        },
        "message": {
          "type": "text"
        },
        "level": {
          "type": "keyword"
        }
      }
    }
  }
}

 

Index template 생성

위에서 생성한 component template을 이용하여 index template을 구성하였습니다.

### create index template
PUT http://localhost:9200/_index_template/my-index-template
Content-Type: application/json

{
  "index_patterns": ["my-index*"],
  "priority": 500,
  "data_stream": {},
  "composed_of": ["my-mappings", "my-settings"]
}

 

Data stream 생성 및 데이터 테스트

위에서 설정한 것을 기반으로 data stream을 생성합니다.

### create data stream
PUT http://localhost:9200/_data_stream/my-index

### data stream 정보 가져오기
GET http://localhost:9200/_data_stream

 

이제 data stream에 실제로 데이터를 넣으면서 확인해보겠습니다.

curl -XPOST http://localhost:9200/my-index/_bulk -H 'Content-Type: application/json' -d '
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068422000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068432000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068242000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068452000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068422000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068432000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068242000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068452000", "level": "error", "message": "failed test message" }

'

 

데이터 확인

data stream으로 생성된 인덱스는. ds의 접두사가 붙어서 인덱스가 생성되었습니다. 이제 생성된 인덱스는 ILM에 설정한 것과 같이 2분 뒤에 Warm 노드로 이동하고 10분 뒤에 삭제됩니다.

2분뒤 .ds-my-index-2023.10.24-000001 인덱스가 shrink-f-pw-.ds-my-index-2023.10.24-000001 인덱스로 변하면서 샤드의 개수가 1개로 줄었습니다.

다시 한번 데이터 인덱싱

curl -XPOST http://localhost:9200/my-index/_bulk -H 'Content-Type: application/json' -d '
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068422000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068432000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068242000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068452000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068422000", "level": "debug", "message": "test message" }
{ "create": {} }
{ "@timestamp": "1698068232000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068432000", "level": "info", "message": "success test message" }
{ "create": {} }
{ "@timestamp": "1698068242000", "level": "error", "message": "failed test message" }
{ "create": {} }
{ "@timestamp": "1698068452000", "level": "error", "message": "failed test message" }

'

 

인덱스의 상태를 확인해 보면 shrink-f-pw-.ds-my-index-2023.10.24-000001에 document가 12개 있고 ds-my-index-2023.10.24-000002에 document가 12개 있는 것을 확인할 수 있습니다.

이제 data stream으로 생성될 수 있는 인덱스는 종류별로 생성되었습니다. data stream으로 다음과 같이 검색을 진행하면 data stream으로 생성된 모든 인덱스에서 검색이 된 것을 확인할 수 있습니다.

### search index
POST http://localhost:9200/my-index/_search
Content-Type: application/json

{
  "size": 1000
}

 

참고

https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-component-template.html

Create or update component template API | Elasticsearch Guide [8.10] | Elastic

https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html

Data streams | Elasticsearch Guide [8.10] | Elastic