Bulk ingest and safe concurrent writes
A bulk request sends several document actions together. Read each item’s status: HTTP 200 can still contain failed items. For an update based on a previous read, use the supported version check to detect another writer’s change.
Prepare your field mapping and check endpoint availability first.
- Prepare the actionsPair each operation with its required document data.
- Send the bulk requestUse the supported action and request format.
- Inspect each item resultSeparate accepted operations from failures.
- Resolve failures carefullyCorrect the cause and reconcile uncertain writes before retrying.
Load many documents
Newline-delimited actions, exactly as a cluster takes them. An index that does not exist yet is created by the first write.
await es.bulk({
body: [
{ index: { _index: 'shop.products', _id: 'sku-1207' } },
{ name: 'Trail 24', brand: 'Aero', price: 89 },
{ index: { _index: 'shop.products', _id: 'sku-3355' } },
{ name: 'City Runner', brand: 'Metro', price: 72 }
]
})Read the result per item
A bulk request is 200 even when some items fail; errors tells you whether any did, and each item carries its own status. Check it — a failed item in a successful request is the classic silent data-loss bug.
const { body: res } = await es.bulk({
body: operations
})
if (res.errors) {
for (const item of res.items) {
const op = item.index ?? item.update ?? item.delete
if (op.status >= 300) console.error(op._id, op.status, op.error?.type)
}
}Guard a write against a concurrent one
Carry if_seq_no and if_primary_term on an action and the write lands only if the document is still the version you read. A stale precondition comes back as a per-item 409 and every other item in the batch still applies, so the loser retries one document rather than the whole load.
// read the document you are about to change
const { body: cur } = await es.get({ index: 'shop.products', id: 'sku-8842' })
await es.bulk({
body: [
{ index: { _index: 'shop.products', _id: 'sku-8842',
if_seq_no: cur._version - 1, if_primary_term: 1 } },
{ ...cur._source, price: 129 }
]
})
// stale -> items[0].index.status === 409
// version_conflict_engine_exceptionwhere preconditions apply
On index actions. A create is already its own precondition (the document must not exist), so it does not take a version. A conditional update or delete is refused: an update is a read-modify-rewrite, and asserting the version at the rewrite is a different guarantee from the one your read saw. External versioning is refused too — it has no faithful mapping onto an exact compare-and-set.
Partial updates
update merges: the fields you send are written and every other field is preserved.
await es.bulk({
body: [
{ update: { _index: 'shop.products', _id: 'sku-8842' } },
{ doc: { price: 139 } } // name, brand, ... untouched
]
})Throughput
Batches are written through the engine's chunked ingest path rather than one document at a time. Send documents in batches of a few hundred to a few thousand; a single enormous request is not faster and costs you the ability to retry a small piece of it.
-
Get the mapping right first — A backfill into the wrong field type means reindexing it.
-
Check what landed — Count and group the data you just loaded.
Elasticsearch is a trademark of Elasticsearch B.V., registered in the U.S. and in other countries. OriginChainDB is not affiliated with, endorsed by, or sponsored by Elasticsearch B.V. OriginChainDB implements a compatible HTTP API so that existing Elasticsearch clients can talk to it; it does not distribute Elasticsearch software.