Elk - Изменение данных

Иногда бывает необходимым поменять уже проиндексированные данные, например, добавить какое-то новое поле. Также может быть вам нужно просто удалить какой-либо документ.

В этой статье поговорим о том какими методами воспользоваться для изменения документов.

Переиндексация (Reindex)

Операция переиндексации - это копия на определенный момент времени. Другими словами, в целевой index будут скопированы только те документы, которые существовали на момент запуска переиндексации. Поэтому, если у вас есть новые документы, проиндексированные после начала операции переиндексации, они не будут приняты во внимание.

Для того чтобы скопировать все документы со старого index в новый используется _reindex. Если при выполнении _reindex у вас не создан конечный (destination) index, то он создастся автоматически. Т.е если у вас схема index не меняется, то нет смысла создавать его вручную.

Пример переноса всех документов с my-index-01 в my-index-02:

PUT my-index-01/_doc/1
{
   "title":"Workspace Paused"
}

PUT my-index-01/_doc/2
{
   "title":"Buy a car"
}

POST _reindex
{
   "source":{
      "index":"my-index-01"
   },
   "dest":{
      "index":"my-index-02"
   }
}

GET my-index-02/_search

Множественная переиндексация

Также используя переиндексацию, вы можете объединять данные из нескольких индексов в один, копируя все документы в новый index. Но тут важно, что если в двух index документы с одинаковыми id и type то в целевом index запишется только один документ. Точнее сперва проиндексируется документ с одного index и потом уже заменится документом другого index.

PUT my-index-01/_doc/1
{
   "title":"Workspace Paused"
}

PUT my-index-02/_doc/4
{
   "title":"Buy a car 2"
}

PUT my-index-01/_doc/4
{
   "title":"Buy a car"
}

POST _reindex
{
   "source":{
      "index": ["my-index-02","my-index-01"]
   },
   "dest":{
      "index":"my-index-03"
   }
}

GET my-index-01/_search

GET my-index-02/_search

GET my-index-03/_search

Используя свойство version_type со значением external вы можете контролировать какое именно значение из нескольких index с одинаковым id проиндексируется. В таком подходе в целевом index окажется документом с наивысшим version.

PUT my-index-01/_doc/1
{
   "title":"Workspace Paused"
}

PUT my-index-02/_doc/4
{
   "title":"Buy a car 2"
}

PUT my-index-02/_doc/4
{
   "title":"Buy a car 2"
}

PUT my-index-02/_doc/4
{
   "title":"Buy a car 2"
}

PUT my-index-01/_doc/4
{
   "title":"Buy a car"
}

POST _reindex
{
   "source":{
      "index": ["my-index-02","my-index-01"]
   },
   "dest":{
      "index":"my-index-03"
   }
}

GET my-index-01/_search

GET my-index-02/_search

GET my-index-03/_search

Переиндексация уникальных документов

Если использовать примеры сверху, то сколько бы вы не запускали операцию _reindex она всегда будет индексировать документы заново. Это происходит потому что по умолчанию работает подход read->delete->ingest.

Для того чтобы индексировать только те документы, что не существуют в целевом (destination) index используется op_type.

POST _reindex
{
   "source":{
      "index": ["my-index-02","my-index-01"]
   },
   "dest":{
      "index":"my-index-03",
      "op_type":"create"
   }
}

Переиндексация с удалённого кластера

Также в качестве source для _reindex можно использовать удалённый (remote) кластер Elasticsearch.

Для использования данного метода придётся задать логин и пароль для подключения к удалённому кластеру.

POST _reindex
{
   "source":{
      "remote":{
         "host":"https://my-remote-cluster:9200",
         "username":"USERNAME",
         "password":"PASSWORD",
         "connect_timeout": "60s"
      }
   },
   "dest":{
      "index":"my-index-03"
   }
}

Throttling

Requests_per_second - параметр ограничивает количество документов, переиндексируемых по секундам. В следующем примере показано, как регулировать операцию переиндексации, ограничивая количество запросов в секунду до 100.

POST _reindex?request_per_second=100
{
   "source":{
      "index":"my-index-01"
   },
   "dest":{
      "index":"my-index-02"
   }
}

Параллельная переиндексация

Можно распараллелить задачу переиндексации с помощью нарезки (Slicing). Распараллеливание может повысить эффективность вашего запроса. Чтобы настроить его, вам нужно определить количество срезов, которое вы хотите использовать. Вы также можете использовать значение auto, чтобы Elasticsearch выбирал сам.

POST _reindex?slices=10
{
   "source":{
      "index":"my-index-01"
   },
   "dest":{
      "index":"my-index-02"
   }
}

Размер партии

Переиндексация происходит партиями (кусками), размер партии по умолчанию - 1000. Также вы можете задать свой значение, используя size.

POST _reindex
{
   "source":{
      "index": "my-index-02",
      "size": 200
   },
   "dest":{
      "index":"my-index-03"
   }
}

Фильтр переиндексации

Бывает так что вам нужно перенести не все документы с index а только какую-то часть. Допустим нужно отсортировать по полю category значение которого является unix.

Для этого используем обычный query:

POST _reindex
{
   "source":{
      "index": "my-index-02",
      "query": {
	    "match": {
	      "category": "unix"
	    }
      }
   },
   "dest":{
      "index":"my-index-03"
   }
}

Изменение данных поля

Когда мы говорили о переиндексации это означало что мы говорим о копировании данных. А что если нам нужно поменять какие-либо данный в index? Elasticsearch предоставляет и такую возможность, для реализации которой придётся использовать _update_by_query.

Например, изменим значение поля title с Buy a car 7 на Don't buy a car в my-index-01.

POST my-index-01/_update_by_query
{
  "script": {
    "source": "ctx._source.title = 'Don't buy a car'",
    "lang": "painless"
  },
  "query": {
    "match_phrase": {
      "title" : "Buy a car 7"
    }
  }
}

GET my-index-01/_search

Если вам нужно поменять значение нескольких полей добавляйте их в поле source разделяя знаком ;, например "source": "ctx._source.title = 'Don't buy a car';ctx._source.category = 'Unix'"

Также _update_by_query используется для обновления структуры (mapping) вашего index. Т.е изменить структуру index вы можете и без этой команды, но нужно также подтвердить изменение index выполнив _update_by_query.

Для примера представим, что у нас есть поле car где описывается марка и модель автомобиля. Нам понадобилось разбить эти значения по отдельности.

1.Добавляем данные в index


PUT my-index-02/_doc/1
{
   "car":"BMW X3"
}

2.Добавляем новое поле


POST my-index-02/_mapping 
{
  "properties": {
    "car": {
      "type": "text",
      "fields": {
        "mark": {
          "type": "keyword"
        }
      }
    }
  }
}

3.Теперь пробуем запустить агрегирование терминов и в итоге получим значение 0


GET my-index-02/_search
{
  "size": 0,
  "aggs": {
    "aggs-demo": {
      "terms": {
        "field": "car.mark"
      }
    }
  }
}

4.А теперь выполним _update_by_query и заново запустим запрос


POST my-index-02/_update_by_query

GET my-index-02/_search
{
  "size": 0,
  "aggs": {
    "aggs-demo": {
      "terms": {
        "field": "car.mark"
      }
    }
  }
}

Удаление документов

Иногда необходимо удалить какой-либо документ из index. Для этого используем _delete_by_query, который нам и позволит это сделать.

Например, если вы хотите удалить все документы с определённым полем title:

POST my-index-02/_delete_by_query
{
  "query": {
    "match_phrase": {
      "title" : "Buy a car 2"
    }
  }  
}

Elk - Изменение данных

О Ynwasg

Комментарии

Рекомендуем к прочтению

Elk - Анализаторы (Analyzers)

ELK - Ingest Node и Pipelines