Реалізація переіндексації даних Elasticsearch без даунтайму
Зміна маппінгу поля — найпоширеніша причина переіндексації. Неможливо змінити тип text на keyword на живому індексі. Неможливо додати новий анализатор до існуючого поля. Рішення: створити новий індекс, перегнати дані через _reindex API, атомарно переключити алиас. Приложення протягом всього процесу продовжує працювати — читає через алиас, який до переключення вказує на старий індекс.
Стратегія blue/green з алиасами
Алиас — абстракція над одним або кількома індексами. Приложення працює з алиасом products, не знаючи фізичної назви індексу.
Початковий стан:
# Перевірити, на що вказує алиас
curl -u elastic:pw "localhost:9200/_alias/products"
# Відповідь:
# { "products_v1": { "aliases": { "products": { "is_write_index": true } } } }
Крок 1 — створити новий індекс з змінним маппінгом:
PUT /products_v2
{
"settings": {
"number_of_shards": 3,
"number_of_replicas": 0,
"refresh_interval": "-1",
"analysis": {
"analyzer": {
"product_analyzer": {
"type": "custom",
"tokenizer": "standard",
"filter": ["lowercase", "russian_stemmer"]
}
}
}
},
"mappings": {
"properties": {
"id": { "type": "keyword" },
"title": {
"type": "text",
"analyzer": "product_analyzer",
"fields": {
"keyword": { "type": "keyword" }
}
},
"price": { "type": "scaled_float", "scaling_factor": 100 },
"new_field": { "type": "keyword" }
}
}
}
number_of_replicas: 0 та refresh_interval: -1 при переіндексації — прискорює завантаження даних.
_reindex API
Крок 2 — запустити переіндексацію:
POST _reindex?wait_for_completion=false
{
"source": {
"index": "products_v1",
"size": 1000
},
"dest": {
"index": "products_v2",
"op_type": "create"
},
"conflicts": "proceed"
}
wait_for_completion=false — завдання йде у фон, повертає task_id. Для великих індексів (>1M документів) обов'язково.
op_type: create — пропустити документ, якщо вже існує (важливо для incremental reindex).
conflicts: proceed — при конфліктах версій продовжити, не перериватися.
Моніторинг прогресу:
# По task_id з відповіді
curl -u elastic:pw "localhost:9200/_tasks/oTUltX4IQMOUUVeiohTt8A:12345?pretty"
# Усі активні reindex завдання
curl -u elastic:pw "localhost:9200/_tasks?actions=*reindex&detailed=true&pretty"
Відповідь завдання містить status.created, status.total — можна рахувати відсоток.
Паралельна переіндексація через slices
Для великих індексів — паралельні slices прискорюють у N разів:
POST _reindex?wait_for_completion=false
{
"source": {
"index": "products_v1",
"size": 500
},
"dest": {
"index": "products_v2"
},
"slices": "auto"
}
slices: auto — автоматично визначає кількість срізів (за кількістю шардів джерела). Кожний slice обробляється паралельно як окреме завдання. Переіндексація 100M документів з 5 шардами при auto йде в 5 потоків.
Проблема: нові документи під час переіндексації
Поки йде reindex, приложення продовжує записувати в products_v1 (через алиас). Нові та оновлені документи не потрапляють у products_v2.
Рішення — incremental sync після основної переіндексації:
POST _reindex?wait_for_completion=false
{
"source": {
"index": "products_v1",
"query": {
"range": {
"updated_at": {
"gte": "2024-01-15T00:00:00",
"lte": "now"
}
}
}
},
"dest": {
"index": "products_v2",
"op_type": "index",
"version_type": "external"
}
}
version_type: external — використовувати _version поле для розв'язання конфліктів. Старі документи не перезапишуть нові.
Для цього в маппінгу повинно бути поле updated_at з датою оновлення. Без нього incremental reindex складний.
Атомарне переключення алиаса
Після завершення переіндексації та incremental sync:
# 1. Відновити production налаштування в новому індексі
PUT /products_v2/_settings
{
"index.number_of_replicas": 1,
"index.refresh_interval": "1s"
}
# 2. Дочекатися відновлення реплік
curl -u elastic:pw "localhost:9200/_cluster/health/products_v2?wait_for_status=green&timeout=30s"
# 3. Атомарно переключити алиас
POST _aliases
{
"actions": [
{
"add": {
"index": "products_v2",
"alias": "products",
"is_write_index": true
}
},
{
"remove": {
"index": "products_v1",
"alias": "products"
}
}
]
}
Операція атомарна — в момент переключення нема стану, коли алиас не вказує на нічого. Запити під час переключення не втрачаються.
Трансформація даних при переіндексації
Reindex підтримує Painless скрипти для трансформації:
POST _reindex
{
"source": {
"index": "products_v1"
},
"dest": {
"index": "products_v2"
},
"script": {
"source": """
// Розділити 'full_name' на 'first_name' та 'last_name'
if (ctx._source.full_name != null) {
def parts = ctx._source.full_name.splitOnToken(' ');
ctx._source.first_name = parts[0];
ctx._source.last_name = parts.length > 1 ? parts[1] : '';
ctx._source.remove('full_name');
}
// Нормалізувати ціну зі строки у число
if (ctx._source.price instanceof String) {
ctx._source.price = Float.parseFloat(ctx._source.price.replace(',', '.'));
}
""",
"lang": "painless"
}
}
План відкату
Якщо після переключення виявлено проблеми — откат за секунди:
POST _aliases
{
"actions": [
{
"add": {
"index": "products_v1",
"alias": "products",
"is_write_index": true
}
},
{
"remove": {
"index": "products_v2",
"alias": "products"
}
}
]
}
Не видаляйте products_v1 відразу — тримайте 24–48 годин для можливості відкату. Потім видаліть, щоб звільнити місце.
Pipeline-переіндексація через ingest
Для збагачення даних при переіндексації — через ingest pipeline:
PUT _ingest/pipeline/enrich-products
{
"processors": [
{
"set": {
"field": "reindexed_at",
"value": "{{_ingest.timestamp}}"
}
},
{
"uppercase": {
"field": "sku",
"ignore_missing": true
}
}
]
}
POST _reindex
{
"source": { "index": "products_v1" },
"dest": {
"index": "products_v2",
"pipeline": "enrich-products"
}
}
Таймлайн
Переіндексація з простою зміною маппінгу — 1 робочий день (планування, запуск, моніторинг, переключення). Складний сценарій з трансформацією даних, інкрементальною синхронізацією та тестуванням — 2–3 дні. Для індексів > 100 млн документів — додатково час на саме виконання reindex (6–24 години залежно від залізо).