[ElasticSearch] 인덱스 매핑 수정하기

엘라스틱서치에서 가장 중요한 작업중 하나가 인덱스의 매핑을 어떻게 설정할 것인가? 라고 생각합니다.

 

기본적으로 동적 매핑을 true로 설정하면 내부적으로 알아서 입력하는 데이터에 맞춰 매핑을 설정하지만, 이렇게 설정되는 매핑이 내가 원하는 설정대로 되지 않는 경우도 종종 발생합니다. 예를 들어 단순 정수값을 데이터로 넣었을 때, 작은 값이더라도 long 타입으로 매핑하기 때문에 비효율적입니다. date 타입 또한 마찬가지로 ISO Date에 맞추지 않은 date 값이 존재하므로 text 타입으로 매핑될 수도 있습니다.

 

그리고 매핑을 수정해야 하는 경우도 이따금씩 발생합니다. 예를 들어, 커스텀 분석기를 수정하거나, 기존 필드에 새로운 서브 필드를 추가해야 하는 경우가 그러합니다.

 

하지만, 엘라스틱 서치의 매핑은 한 번 설정했으면 그 다음부터 수정이 불가합니다. 이미 설정한 필드는 그 매핑을 유지하게 됩니다.

 

따라서, 인덱스를 매핑하기 위해서는 새로운 인덱스를 생성하고 해당 인덱스로 데이터를 옮긴 후에 alias(별칭)을 바꾸는 방법을 통해 인덱스 매핑을 수정해야 합니다.

 

향후 예제들은 Node.js 환경에서 @elastic/elasticsearch 패키지를 사용했습니다.

1. 인덱스 생성

인덱스 매핑을 수정한 새로운 인덱스를 생성합니다.

const client = new Client({
  cloud: {
    id: "cloudId",
  },
  auth: {
    apiKey: "apiKey",
  },
});

async function run() {
    await client.indices.create({
    index: "store_240927",
    settings: {
      max_ngram_diff: 15,
      analysis: {
        analyzer: {
          kr_nori: {
            type: "nori",
          },
          jp_kuromoji: {
            type: "kuromoji",
          },
          en_english: {
            type: "english",
          },
          name_ngram_tokenizer: {
            type: "custom",
            tokenizer: "name_ngram_tokenizer",
          },
          tag_ngram_tokenizer: {
            type: "custom",
            tokenizer: "tag_ngram_tokenizer",
          },
        },
        tokenizer: {
          name_ngram_tokenizer: {
            type: "ngram",
            min_gram: 1,
            max_gram: 12,
          },
          tag_ngram_tokenizer: {
            type: "ngram",
            min_gram: 1,
            max_gram: 15,
          },
        },
      },
    },
    mappings: {
      properties: {
        createdAt: {
          type: "date",
          format: "strict_date_time || epoch_millis",
        },
        description: {
          type: "text",
          fields: {
            kr: {
              type: "text",
              analyzer: "kr_nori",
            },
            jp: {
              type: "text",
              analyzer: "jp_kuromoji",
            },
            en: {
              type: "text",
              analyzer: "en_english",
            },
          },
        },
        name: {
          type: "text",
          analyzer: "name_ngram_tokenizer",
          fields: {
            completion: {
              type: "completion",
            },
          },
        },
        updatedAt: {
          type: "date",
          format: "strict_date_time || epoch_millis",
        },
      },
    },
  });
}

run().catch(console.log);

예를 들어 위처럼 name 필드를 text 타입으로 설정하였는데, keyword 서브 필드를 추가해야 하는 상황이라고 한다면, 새로운 인덱스를 먼저 생성합니다.

await client.indices.create({
    index: "store_240928",
    settings: {
      max_ngram_diff: 15,
      analysis: {
        analyzer: {
          kr_nori: {
            type: "nori",
          },
          jp_kuromoji: {
            type: "kuromoji",
          },
          en_english: {
            type: "english",
          },
          name_ngram_tokenizer: {
            type: "custom",
            tokenizer: "name_ngram_tokenizer",
          },
          tag_ngram_tokenizer: {
            type: "custom",
            tokenizer: "tag_ngram_tokenizer",
          },
        },
        tokenizer: {
          name_ngram_tokenizer: {
            type: "ngram",
            min_gram: 1,
            max_gram: 12,
          },
          tag_ngram_tokenizer: {
            type: "ngram",
            min_gram: 1,
            max_gram: 15,
          },
        },
      },
    },
    mappings: {
      properties: {
        createdAt: {
          type: "date",
          format: "strict_date_time || epoch_millis",
        },
        description: {
          type: "text",
          fields: {
            kr: {
              type: "text",
              analyzer: "kr_nori",
            },
            jp: {
              type: "text",
              analyzer: "jp_kuromoji",
            },
            en: {
              type: "text",
              analyzer: "en_english",
            },
          },
        },
        name: {
          type: "text",
          analyzer: "name_ngram_tokenizer",
          fields: {
            completion: {
              type: "completion",
            },
            keyword: {
              type: "keyword",
            },
          },
        },
        updatedAt: {
          type: "date",
          format: "strict_date_time || epoch_millis",
        },
      },
    },
  });

2. 리인덱싱

특정 인덱스에 존재하는 데이터를 다른 인덱스로 복사하고 색인하는 과정을 리인덱싱이라고 부릅니다. 복잡한 과정 없이 데이터를 일괄 수행할 수 있어 매우 유용한 api입니다.

  await client.reindex({
    source: {
      index: "store_240927",
    },
    dest: {
      index: "store_240928",
    },
  });

source에는 색인된 데이터가 저장된 인덱스를, dest에는 해당 데이터를 옮길 인덱스를 넣으면 됩니다.

 

그런데 리인덱싱 과정은 모든 데이터를 옮겨야 하므로 작은 작업이 아닙니다.

따라서 timeout이 발생할 수 있습니다. 이러한 경우에는 따로 timeout 설정을 조정하거나, 비동기로 실행해주어야 합니다.

1) timeout 설정

timeout은 elastic search client와 연결할 때 설정하는 requestTimeout과, reindex API의 timeout이 있습니다. 저의 경우에는 reindex 시간이 1분을 약간 넘어서 기본값을 넘어가므로 별도로 timeout을 설정해주어야 했습니다.

const client = new Client({
  cloud: {
    id: "clientId",
  },
  auth: {
    apiKey: "apiKey",
  },
  requestTimeout: "5m",
});

 

마이그레이션을 위한 작업이므로 timeout을 길게 가져갔지만, 그 외에 실무에서 사용하는 것은 좋지 않습니다.

2) 비동기 수행

리인덱싱을 기다리지 않고 비동기적으로 수행하는 방법도 있습니다.

await client.reindex({
  source: {
    index: "store_240927",
  },
  dest: {
    index: "store_240928",
  },
  wait_for_completion: false
});

 

이렇게 wait_for_completion을 false로 하면 리인덱스를 비동기적으로 수행할 수 있습니다. 이때의 반환값은 task의 id가 됩니다.

"tasks" : {
  "oTUltX4IQMOUUVeiohTt8A:124" : {
    "node" : "oTUltX4IQMOUUVeiohTt8A",
    "id" : 124,
    "type" : "direct",
    "action" : "cluster:monitor/tasks/lists[n]",
    "start_time_in_millis" : 1458585884904,
    "running_time_in_nanos" : 47402,
    "cancellable" : false,
    "parent_task_id" : "oTUltX4IQMOUUVeiohTt8A:123"
  },
}

 

이때 task의 현재 상황을 파악하기 위해 get task API를 통해 조회할 수 있습니다.

GET _tasks/oTUltX4IQMOUUVeiohTt8A:12345

await client.tasks.get({
  id: "store_240928:1",
})

 

그리고 get task API도 해당 task가 끝날 때까지 대기할 수 있습니다.

curl -X GET "localhost:9200/_tasks/oTUltX4IQMOUUVeiohTt8A:12345?wait_for_completion=true&timeout=10s&pretty"

 

혹은 리인덱싱 태스크를 모두 확인할 수 있습니다.

curl -X GET "localhost:9200/_tasks?actions=*reindex&wait_for_completion=true&timeout=10s&pretty"

 

개인적으로 굳이 비동기적으로 가져갈 필요가 없어서 timeout을 늘려서 리인덱스를 완료했습니다.

 

3. 인덱스 alias(별칭) 변경

마지막으로 인덱스의 별칭을 변경해주어야 합니다.

엘라스틱서치 API를 사용할 때는 인덱스를 명시하거나 혹은 해당 인덱스의 별칭을 사용하여 API를 사용합니다. 그래서 단순히 별칭을 스와핑하여 원래 사용하던 인덱스에서 다른 인덱스로 별칭을 전환하여 중단 없이 API 사용이 가능해집니다.

await client.indices.updateAliases({
  actions: [
    { remove: { index: "store_240927", alias: "store" } },
    { add: { index: "store_240928", alias: "store" } }
  ],
});

 

이렇게 alias를 한 번에 업데이트하면 원래 store_240927 인덱스로 흘러가던 요청이 store_240928 인덱스로 흘러갈 수 있게 됩니다.

 

하지만 이때 문제가 하나 있는데, alias로 사용하려는 텍스트와 같은 텍스트를 사용하는 index가 있으면 안된다는 것입니다.

예를 들어, 위에서 store라고 alias를 설정하였는데 이미 "store" 인덱스가 있는 경우. alias를 추가하거나 삭제할 수 없습니다.

 

이를 해결하기 위해서는 해당 인덱스를 삭제해주어야 합니다.

await client.delete({
  index: "store",
})

await client.indices.updateAliases({
  actions: [
    { add: { index: "store_240928", alias: "store" } }],
});

 

리인덱스 사용 시 고려하면 좋은 것들

 

리인덱싱을 사용할 때 고려할 점은 적지 않은 것 같습니다.

 

먼저 reindex는 검색 후 bulk로 추가되는 구조이므로, 해당 bulk의 크기를 어떻게 가져갈 것인지가 성능에 영향을 미칠 수 있습니다. 이때는 리인덱스 API의 파라미터를 조정하여 bulk 사이즈를 증감시켜 최적의 파라미터를 찾는 것이 좋습니다.

 

그리고 리인덱스 중간에는 해당 인덱스로 검색 요청이 들어올 일이 없으므로, 색인을 위해 refresh하는 간격을 없애는 것도 좋습니다. 그래서 refresh_interval은 -1로 설정하면 좋습니다.

 

 

참고/출처

• 다나와 반윤성님의 Elasticsearch 3TB의 인덱스를 reindex 하는 방법 -  https://danawalab.github.io/elastic/2023/05/16/elasticReindex.html

Elasticsearch 3TB의 인덱스를 reindex 하는 방법

• es 공식문서 task management API - https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html#tasks-api-response-codes

Task management API | Elasticsearch Guide [8.15] | Elastic