API обновления по запросу

Обновление по запросу поддерживает скрипты для обновления источника документа. Например, следующий запрос увеличивает поле count для всех документов со значением user.id равным kimchy в my-index-000001:

POST my-index-000001/_update_by_query
{
  "script": {
    "source": "ctx._source.count++",
    "lang": "painless"
  },
  "query": {
    "term": {
      "user.id": "kimchy"
    }
  }
}

Обратите внимание, что conflicts=proceed не указано в этом примере. В этом случае конфликт версий должен остановить процесс, чтобы вы могли обработать ошибку.

Как и в API обновления, вы можете установить ctx.op, чтобы изменить выполняемую операцию:

Обновление по запросу поддерживает только update, noop и delete. Установка ctx.op на другое значение является ошибкой. Установка любого другого поля в ctx является ошибкой. Этот API позволяет только изменять исходные данные совпадающих документов, вы не можете их перемещать.

Обновление по запросу может использовать функцию конвейеров ingest, указав pipeline:

PUT _ingest/pipeline/set-foo
{
  "description" : "sets foo",
  "processors" : [ {
      "set" : {
        "field": "foo",
        "value": "bar"
      }
  } ]
}
POST my-index-000001/_update_by_query?pipeline=set-foo

Вы можете получить статус всех выполняемых запросов обновления по запросу с помощью API управления задачами:

GET _tasks?detailed=true&actions=*byquery

Ответы выглядят следующим образом:

{
  "nodes" : {
    "r1A2WoRbTwKZ516z6NEs5A" : {
      "name" : "r1A2WoR",
      "transport_address" : "127.0.0.1:9300",
      "host" : "127.0.0.1",
      "ip" : "127.0.0.1:9300",
      "attributes" : {
        "testattr" : "test",
        "portsfile" : "true"
      },
      "tasks" : {
        "r1A2WoRbTwKZ516z6NEs5A:36619" : {
          "node" : "r1A2WoRbTwKZ516z6NEs5A",
          "id" : 36619,
          "type" : "transport",
          "action" : "indices:data/write/update/byquery",
          "status" : {    
            "total" : 6154,
            "updated" : 3500,
            "created" : 0,
            "deleted" : 0,
            "batches" : 4,
            "version_conflicts" : 0,
            "noops" : 0,
            "retries": {
              "bulk": 0,
              "search": 0
            },
            "throttled_millis": 0
          },
          "description" : ""
        }
      }
    }
  }
}

С помощью идентификатора задачи вы можете получить доступ к задаче напрямую. Следующий пример извлекает информацию о задаче r1A2WoRbTwKZ516z6NEs5A:36619:

GET /_tasks/r1A2WoRbTwKZ516z6NEs5A:36619

Преимущество этого API заключается в том, что он интегрируется с wait_for_completion=false, чтобы прозрачно возвращать состояние завершенных задач. Если задача завершена и для неё было установлено wait_for_completion=false, она вернётся с полем results или error. Стоимостью этой функции является документ, который wait_for_completion=false создаёт в .tasks/task/${taskId}. Вам необходимо удалить этот документ самостоятельно.

Любую операцию обновления по запросу можно отменить, используя API отмены задач:

POST _tasks/r1A2WoRbTwKZ516z6NEs5A:36619/_cancel

Идентификатор задачи можно найти, используя API задач.

Отмена должна происходить быстро, но может занять несколько секунд. API статуса задач выше будет продолжать отображать задачу обновления по запросу до тех пор, пока эта задача не проверит, что она была отменена, и не завершит себя.

Значение requests_per_second можно изменить на выполняющемся обновлении по запросу, используя API _rethrottle:

POST _update_by_query/r1A2WoRbTwKZ516z6NEs5A:36619/_rethrottle?requests_per_second=-1

Идентификатор задачи можно найти, используя API задач.

Как и при установке значения с помощью API _update_by_query, значение requests_per_second может быть установлено в -1, чтобы отключить ограничение скорости, или в любое десятичное число, например, 1.7 или 12, чтобы установить ограничение на этом уровне. Изменение ограничения скорости, ускоряющее запрос, вступает в силу немедленно, но изменение, замедляющее запрос, вступит в силу после завершения текущей партии. Это предотвращает таймауты при прокрутке.

Разбейте запрос обновления по запросу на части вручную, указав идентификатор части и общее количество частей в каждом запросе:

POST my-index-000001/_update_by_query
{
  "slice": {
    "id": 0,
    "max": 2
  },
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}
POST my-index-000001/_update_by_query
{
  "slice": {
    "id": 1,
    "max": 2
  },
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}

Что можно проверить с помощью:

GET _refresh
POST my-index-000001/_search?size=0&q=extra:test&filter_path=hits.total

Что приводит к осмысленному total, например, такому:

{
  "hits": {
    "total": {
        "value": 120,
        "relation": "eq"
    }
  }
}

Вы также можете позволить обновлению по запросу автоматически распараллеливать использование скроллирования с нарезкой для нарезки по _id. Используйте slices, чтобы указать количество частей:

POST my-index-000001/_update_by_query?refresh&slices=5
{
  "script": {
    "source": "ctx._source['extra'] = 'test'"
  }
}

Что также можно проверить с помощью:

POST my-index-000001/_search?size=0&q=extra:test&filter_path=hits.total

Что приводит к осмысленному total, например, такому:

{
  "hits": {
    "total": {
        "value": 120,
        "relation": "eq"
    }
  }
}

Установка значения slices в auto позволит Elasticsearch выбрать количество используемых частей. Эта настройка будет использовать по одной части на фрагмент, до определенного предела. Если есть несколько потоков данных или индексов, количество частей будет выбираться на основе индекса или базового индекса с наименьшим количеством фрагментов.

Добавление slices к _update_by_query просто автоматизирует ручной процесс, описанный выше, создавая подзапросы, что означает, что у него есть некоторые особенности:

Предположим, вы создали индекс без динамического отображения, заполнили его данными, а затем добавили значение отображения, чтобы получить больше полей из данных:

PUT test
{
  "mappings": {
    "dynamic": false,   
    "properties": {
      "text": {"type": "text"}
    }
  }
}

POST test/_doc?refresh
{
  "text": "words words",
  "flag": "bar"
}
POST test/_doc?refresh
{
  "text": "words words",
  "flag": "foo"
}
PUT test/_mapping   
{
  "properties": {
    "text": {"type": "text"},
    "flag": {"type": "text", "analyzer": "keyword"}
  }
}

Поиск данных ничего не найдет:

POST test/_search?filter_path=hits.total
{
  "query": {
    "match": {
      "flag": "foo"
    }
  }
}

{
  "hits" : {
    "total": {
        "value": 0,
        "relation": "eq"
    }
  }
}

Но вы можете выполнить запрос _update_by_query, чтобы получить новое отображение:

POST test/_update_by_query?refresh&conflicts=proceed
POST test/_search?filter_path=hits.total
{
  "query": {
    "match": {
      "flag": "foo"
    }
  }
}

{
  "hits" : {
    "total": {
        "value": 1,
        "relation": "eq"
    }
  }
}

Вы можете сделать то же самое при добавлении поля в многопольное поле.