API массовой обработки

POST /_bulk

POST /<target>/_bulk

Предоставляет способ выполнения нескольких действий index, create, delete и update в одном запросе.

Действия задаются в теле запроса с использованием структуры JSON с разделителями строк (NDJSON):

action_and_meta_data\n
optional_source\n
action_and_meta_data\n
optional_source\n
....
action_and_meta_data\n
optional_source\n

Действия index и create ожидают исходный документ в следующей строке и имеют ту же семантику, что и параметр op_type в стандартном API индексирования: create завершается ошибкой, если документ с таким же идентификатором уже существует в целевом индексе; index добавляет или заменяет документ по необходимости.

update ожидает, что частичный документ, upsert, скрипт и его параметры будут указаны в следующей строке.

delete не ожидает исходного документа в следующей строке и имеет ту же семантику, что и стандартный API удаления.

Поскольку этот формат использует литеральные символы \n в качестве разделителей, убедитесь, что JSON-действия и исходные данные не отформатированы с помощью отступов.

Если вы предоставляете <target> в пути запроса, он используется для любых действий, которые явно не указывают аргумент _index.

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

Клиентские библиотеки, использующие этот протокол, должны пытаться сделать нечто подобное на стороне клиента и минимизировать буферизацию.

Нет «правильного» количества действий, которые следует выполнять в одном запросе массовой обработки. Экспериментируйте с различными настройками, чтобы найти оптимальный размер для вашей конкретной рабочей нагрузки. Обратите внимание, что Elasticsearch по умолчанию ограничивает максимальный размер HTTP-запроса значением 100mb, поэтому клиенты должны гарантировать, что ни один запрос не превышает этот размер. Индексирование одного документа, размер которого превышает предел, невозможно, поэтому необходимо предварительно обработать такие документы, разделив их на меньшие части, прежде чем отправлять их в Elasticsearch. Например, разделите документы на страницы или главы перед индексированием или храните исходные бинарные данные в системе вне Elasticsearch, заменяя исходные данные ссылкой на внешнюю систему в документах, которые вы отправляете в Elasticsearch.

Некоторые из официально поддерживаемых клиентов предоставляют вспомогательные средства для работы с запросами массовой обработки и повторной индексацией:

Если вы предоставляете текстовый файл в качестве входных данных для curl, вы обязаны использовать флаг --data-binary вместо простого -d. Последний вариант не сохраняет символы новой строки. Пример:

$ cat requests
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
$ curl -s -H "Content-Type: application/x-ndjson" -XPOST localhost:9200/_bulk --data-binary "@requests"; echo
{"took":7, "errors": false, "items":[{"index":{"_index":"test","_type":"_doc","_id":"1","_version":1,"result":"created","forced_refresh":false}}]}

Каждое действие index и delete в вызове API массовой обработки может включать параметры if_seq_no и if_primary_term в соответствующих строках действия и метаданных. Параметры if_seq_no и if_primary_term управляют выполнением операций, в зависимости от последнего изменения существующих документов. См. Оптимистический контроль конкурентности для получения дополнительной информации.

Каждый элемент массовой обработки может включать значение версии с помощью поля version. Оно автоматически следует поведению операции индексирования/удаления на основе сопоставления _version. Также поддерживается version_type (см. версионирование).

Каждый элемент массовой обработки может включать значение маршрутизации с помощью поля routing. Оно автоматически следует поведению операции индексирования/удаления на основе сопоставления _routing.

При выполнении запросов массовой обработки вы можете установить параметр wait_for_active_shards для требования о минимальном количестве активных копий фрагментов перед началом обработки запроса массовой обработки. См. здесь для получения дополнительной информации и примера использования.

Управление моментом, когда изменения, внесенные этим запросом, будут доступны для поиска. См. refresh.

См. Управление доступом на основе URL.

Тело запроса содержит список действий create, delete, index и update, разделённых символом новой строки, и связанные с ними исходные данные.

POST _bulk
{ "index" : { "_index" : "test", "_id" : "1" } }
{ "field1" : "value1" }
{ "delete" : { "_index" : "test", "_id" : "2" } }
{ "create" : { "_index" : "test", "_id" : "3" } }
{ "field1" : "value3" }
{ "update" : {"_id" : "1", "_index" : "test"} }
{ "doc" : {"field2" : "value2"} }

API возвращает следующий результат:

{
   "took": 30,
   "errors": false,
   "items": [
      {
         "index": {
            "_index": "test",
            "_type": "_doc",
            "_id": "1",
            "_version": 1,
            "result": "created",
            "_shards": {
               "total": 2,
               "successful": 1,
               "failed": 0
            },
            "status": 201,
            "_seq_no" : 0,
            "_primary_term": 1
         }
      },
      {
         "delete": {
            "_index": "test",
            "_type": "_doc",
            "_id": "2",
            "_version": 1,
            "result": "not_found",
            "_shards": {
               "total": 2,
               "successful": 1,
               "failed": 0
            },
            "status": 404,
            "_seq_no" : 1,
            "_primary_term" : 2
         }
      },
      {
         "create": {
            "_index": "test",
            "_type": "_doc",
            "_id": "3",
            "_version": 1,
            "result": "created",
            "_shards": {
               "total": 2,
               "successful": 1,
               "failed": 0
            },
            "status": 201,
            "_seq_no" : 2,
            "_primary_term" : 3
         }
      },
      {
         "update": {
            "_index": "test",
            "_type": "_doc",
            "_id": "1",
            "_version": 2,
            "result": "updated",
            "_shards": {
                "total": 2,
                "successful": 1,
                "failed": 0
            },
            "status": 200,
            "_seq_no" : 3,
            "_primary_term" : 4
         }
      }
   ]
}

При использовании действия update, retry_on_conflict может использоваться в качестве поля самого действия (а не в дополнительной строке полезной нагрузки), чтобы указать, сколько раз следует повторить попытку обновления в случае конфликта версий.

Полезная нагрузка действия update поддерживает следующие опции: doc (частичный документ), upsert, doc_as_upsert, script, params (для скрипта), lang (для скрипта) и _source. Подробности об опциях см. в документации по обновлению. Пример с действиями обновления:

POST _bulk
{ "update" : {"_id" : "1", "_index" : "index1", "retry_on_conflict" : 3} }
{ "doc" : {"field" : "value"} }
{ "update" : { "_id" : "0", "_index" : "index1", "retry_on_conflict" : 3} }
{ "script" : { "source": "ctx._source.counter += params.param1", "lang" : "painless", "params" : {"param1" : 1}}, "upsert" : {"counter" : 1}}
{ "update" : {"_id" : "2", "_index" : "index1", "retry_on_conflict" : 3} }
{ "doc" : {"field" : "value"}, "doc_as_upsert" : true }
{ "update" : {"_id" : "3", "_index" : "index1", "_source" : true} }
{ "doc" : {"field" : "value"} }
{ "update" : {"_id" : "4", "_index" : "index1"} }
{ "doc" : {"field" : "value"}, "_source": true}

Следующий пакетный запрос API включает операции, которые обновляют несуществующие документы.

POST /_bulk
{ "update": {"_id": "5", "_index": "index1"} }
{ "doc": {"my_field": "foo"} }
{ "update": {"_id": "6", "_index": "index1"} }
{ "doc": {"my_field": "foo"} }
{ "create": {"_id": "7", "_index": "index1"} }
{ "my_field": "foo" }

Поскольку эти операции не могут завершиться успешно, API возвращает ответ со флагом errors, установленным в значение true.

Ответ также содержит объект error для любых неудачных операций. Объект error содержит дополнительную информацию об ошибке, например, тип ошибки и причину.

{
  "took": 486,
  "errors": true,
  "items": [
    {
      "update": {
        "_index": "index1",
        "_type" : "_doc",
        "_id": "5",
        "status": 404,
        "error": {
          "type": "document_missing_exception",
          "reason": "[_doc][5]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "update": {
        "_index": "index1",
        "_type" : "_doc",
        "_id": "6",
        "status": 404,
        "error": {
          "type": "document_missing_exception",
          "reason": "[_doc][6]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "create": {
        "_index": "index1",
        "_type" : "_doc",
        "_id": "7",
        "_version": 1,
        "result": "created",
        "_shards": {
          "total": 2,
          "successful": 1,
          "failed": 0
        },
        "_seq_no": 0,
        "_primary_term": 1,
        "status": 201
      }
    }
  ]
}

Чтобы вернуть только информацию об ошибочных операциях, используйте параметр запроса filter_path со значением items.*.error.

POST /_bulk?filter_path=items.*.error
{ "update": {"_id": "5", "_index": "index1"} }
{ "doc": {"my_field": "baz"} }
{ "update": {"_id": "6", "_index": "index1"} }
{ "doc": {"my_field": "baz"} }
{ "update": {"_id": "7", "_index": "index1"} }
{ "doc": {"my_field": "baz"} }

API возвращает следующий результат.

{
  "items": [
    {
      "update": {
        "error": {
          "type": "document_missing_exception",
          "reason": "[_doc][5]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "update": {
        "error": {
          "type": "document_missing_exception",
          "reason": "[_doc][6]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    }
  ]
}

В приведенном ниже примере создается динамическая шаблон, а затем выполняется пакетный запрос, состоящий из запросов index/create с параметром dynamic_templates.

PUT my-index/
{
  "mappings": {
    "dynamic_templates": [
      {
        "geo_point": {
             "mapping": {
                "type" : "geo_point"
             }
        }
      }
    ]
  }
}

POST /_bulk
{ "index" : { "_index" : "my_index", "_id" : "1", "dynamic_templates": {"work_location": "geo_point"}} }
{ "field" : "value1", "work_location": "41.12,-71.34", "raw_location": "41.12,-71.34"}
{ "create" : { "_index" : "my_index", "_id" : "2", "dynamic_templates": {"home_location": "geo_point"}} }
{ "field" : "value2", "home_location": "41.12,-71.34"}

Пакетный запрос создает два новых поля work_location и home_location типа geo_point в соответствии с параметром dynamic_templates; однако поле raw_location создается с помощью стандартных правил динамического отображения, в данном случае как поле типа text, поскольку оно передается как строка в JSON-документе.