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 возвращает ошибку, если документ с тем же ID уже существует в целевом индексе, 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","_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 и связанных с ними исходных данных.

$params = [
    'body' => [
        [
            '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',
            ],
        ],
    ],
];
$response = $client->bulk($params);

resp = client.bulk(
    operations=[
        {
            "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"
            }
        }
    ],
)
print(resp)

response = client.bulk(
  body: [
    {
      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'
      }
    }
  ]
)
puts response

res, err := es.Bulk(
	strings.NewReader(`
{ "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"} }
`),
)
fmt.Println(res, err)

const response = await client.bulk({
  operations: [
    {
      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",
      },
    },
  ],
});
console.log(response);

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",
            "_id": "1",
            "_version": 1,
            "result": "created",
            "_shards": {
               "total": 2,
               "successful": 1,
               "failed": 0
            },
            "status": 201,
            "_seq_no" : 0,
            "_primary_term": 1
         }
      },
      {
         "delete": {
            "_index": "test",
            "_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",
            "_id": "3",
            "_version": 1,
            "result": "created",
            "_shards": {
               "total": 2,
               "successful": 1,
               "failed": 0
            },
            "status": 201,
            "_seq_no" : 2,
            "_primary_term" : 3
         }
      },
      {
         "update": {
            "_index": "test",
            "_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. Подробности об опциях см. в документации по обновлению. Пример с действиями обновления:

$params = [
    'body' => [
        [
            '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,
        ],
    ],
];
$response = $client->bulk($params);

resp = client.bulk(
    operations=[
        {
            "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
        }
    ],
)
print(resp)

response = client.bulk(
  body: [
    {
      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
    }
  ]
)
puts response

res, err := es.Bulk(
	strings.NewReader(`
{ "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}
`),
)
fmt.Println(res, err)

const response = await client.bulk({
  operations: [
    {
      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,
    },
  ],
});
console.log(response);

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 включает операции обновления несуществующих документов.

$params = [
    'body' => [
        [
            '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',
        ],
    ],
];
$response = $client->bulk($params);

resp = client.bulk(
    operations=[
        {
            "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"
        }
    ],
)
print(resp)

response = client.bulk(
  body: [
    {
      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'
    }
  ]
)
puts response

res, err := es.Bulk(
	strings.NewReader(`
{ "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" }
`),
)
fmt.Println(res, err)

const response = await client.bulk({
  operations: [
    {
      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",
    },
  ],
});
console.log(response);

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",
        "_id": "5",
        "status": 404,
        "error": {
          "type": "document_missing_exception",
          "reason": "[5]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "update": {
        "_index": "index1",
        "_id": "6",
        "status": 404,
        "error": {
          "type": "document_missing_exception",
          "reason": "[6]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "create": {
        "_index": "index1",
        "_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.

$params = [
    'body' => [
        [
            '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',
            ],
        ],
    ],
];
$response = $client->bulk($params);

resp = client.bulk(
    filter_path="items.*.error",
    operations=[
        {
            "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"
            }
        }
    ],
)
print(resp)

response = client.bulk(
  filter_path: 'items.*.error',
  body: [
    {
      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'
      }
    }
  ]
)
puts response

res, err := es.Bulk(
	strings.NewReader(`
{ "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"} }
`),
	es.Bulk.WithFilterPath("items.*.error"),
)
fmt.Println(res, err)

const response = await client.bulk({
  filter_path: "items.*.error",
  operations: [
    {
      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",
      },
    },
  ],
});
console.log(response);

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": "[5]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    },
    {
      "update": {
        "error": {
          "type": "document_missing_exception",
          "reason": "[6]: document missing",
          "index_uuid": "aAsFqTI0Tc2W0LCWgPNrOA",
          "shard": "0",
          "index": "index1"
        }
      }
    }
  ]
}

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

resp = client.indices.create(
    index="my-index",
    mappings={
        "dynamic_templates": [
            {
                "geo_point": {
                    "mapping": {
                        "type": "geo_point"
                    }
                }
            }
        ]
    },
)
print(resp)

resp1 = client.bulk(
    operations=[
        {
            "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"
        }
    ],
)
print(resp1)

response = client.indices.create(
  index: 'my-index',
  body: {
    mappings: {
      dynamic_templates: [
        {
          geo_point: {
            mapping: {
              type: 'geo_point'
            }
          }
        }
      ]
    }
  }
)
puts response

response = client.bulk(
  body: [
    {
      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'
    }
  ]
)
puts response

const response = await client.indices.create({
  index: "my-index",
  mappings: {
    dynamic_templates: [
      {
        geo_point: {
          mapping: {
            type: "geo_point",
          },
        },
      },
    ],
  },
});
console.log(response);

const response1 = await client.bulk({
  operations: [
    {
      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",
    },
  ],
});
console.log(response1);

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-документе.