Batch operations

Applies to

• Aerospike Developer SDK preview (Java 21+ and Python 3.10+)
• Aerospike Database 6.0 or later unless a section states otherwise

Code examples

Examples assume a connected session from Connect to Aerospike. Each usage guide lists shared imports in a tabbed block near the top of the page; snippet blocks add imports only for types not already shown. The first code block in each section includes dataset setup; later blocks in that section reuse those values. When a page includes a Complete example section, that block is fully self-contained.

Learn how to perform multiple database operations efficiently in a single network request. Batch operations reduce latency and improve throughput when working with multiple records.

Except where noted, snippets on this page use the imports below. A snippet lists additional import lines only when it needs a type not shown here. When this page includes a Complete example section, that block is fully self-contained with every import required to run it.

Java

import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Record;
import com.aerospike.client.sdk.RecordResult;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.ErrorStrategy;

import java.util.List;

Python

from aerospike_sdk import DataSet

Why use batch operations?

Approach	100 Records	Network Roundtrips
Individual operations	~100ms+	100
Batch operation	~5-10ms	1

Batch operations send multiple requests in a single network call, as well as sending requests to multiple servers concurrently, dramatically reducing latency.

Batch read

Read multiple records in one request:

Java

DataSet users = DataSet.of("test", "users");

RecordStream result = session.query(users.ids("user-1", "user-2", "user-3"))
    .execute();

// Get all records
List<Record> records = new java.util.ArrayList<>();
result.forEach(rr -> {
    if (rr.isOk()) {
        records.add(rr.recordOrThrow());
    }
});
for (Record record : records) {
    System.out.println("Name: " + record.getString("name"));
}

// Or access by index
Record first = records.get(0);  // user-1
Record second = records.get(1); // user-2

📖 API reference: DataSet.of(...) | DataSet.ids(...) | Session.query(List) | ChainableQueryBuilder.execute() | RecordStream.forEach(...) | RecordResult.isOk() | RecordResult.recordOrThrow() | Record.getString(...)

Python

users = DataSet.of("test", "users")

stream = await session.query(users.ids("user-1", "user-2", "user-3")).execute()

# Get all records
records = []
async for rr in stream:
    if rr.is_ok:
        records.append(rr.record_or_raise())
stream.close()

for record in records:
    print(f"Name: {record.bins['name']}")

# Or access by index
first = records[0]   # user-1
second = records[1]  # user-2

📖 API reference: DataSet.of() | DataSet.ids() | Session.query() | QueryBuilder.execute() | RecordResult.is_ok | RecordResult.record_or_raise() | RecordStream.close()

Batch write

Create or update multiple records:

Java

DataSet users = DataSet.of("test", "users");

session.insert(users)
    .bins("name", "email")
    .id("user-1").values("Alice", "alice@example.com")
    .id("user-2").values("Bob", "bob@example.com")
    .id("user-3").values("Carol", "carol@example.com")
    .execute();

📖 API reference: Session.insert(DataSet) | OperationObjectBuilder.bins(...) | IdValuesBuilder.id(...) | IdValuesRowBuilder.values(...) | ChainableQueryBuilder.execute()

Python

users = DataSet.of("test", "users")

await (
    session.batch()
    .insert(users.id("user-1")).put({"name": "Alice", "email": "alice@example.com"})
    .insert(users.id("user-2")).put({"name": "Bob",   "email": "bob@example.com"})
    .insert(users.id("user-3")).put({"name": "Carol", "email": "carol@example.com"})
    .execute()
)

📖 API reference: DataSet.of() | DataSet.id() | Session.batch() | BatchOperationBuilder.insert() | BatchKeyOperationBuilder.put() | BatchOperationBuilder.execute()

Integer keys

DataSet.id() accepts any Python value, string or integer. Pass an int directly when you need integer keys (such as users.id(42)), not users.id(str(42)). Integer and string keys are distinct in Aerospike, so reads using integer keys will not find records written with string keys and vice versa.

# String key           → Aerospike key type: String
users.id("42")

# Integer key          → Aerospike key type: Integer
users.id(42)

When looping over a numeric range, pass the integer directly:

leaderboard = DataSet.of("test", "leaderboard")
batch = session.batch()
for i in range(1, 51):
    batch.upsert(leaderboard.id(i)).put({"name": f"Player {i}", "score": i * 100})
await batch.execute()

Batch upsert

Insert or update multiple records:

Java

DataSet users = DataSet.of("test", "users");

session
    .upsert(users.ids("user-1","user-2"))
        .bin("status").setTo("active")
        .bin("balance").add(500)
    .upsert(users.id("user-3"))
        .bin("status").setTo("inactive")
    .execute();

📖 API reference: DataSet.ids(...) | DataSet.id(...) | ChainableQueryBuilder.bin(...) | BinBuilder.add(...) | ChainableQueryBuilder.execute()

Python

users = DataSet.of("test", "users")

await (
    session.batch()
    .upsert(users.id("user-1")).bin("status").set_to("active").bin("balance").add(500)
    .upsert(users.id("user-2")).bin("status").set_to("active").bin("balance").add(500)
    .upsert(users.id("user-3")).bin("status").set_to("inactive")
    .execute()
)

📖 API reference: DataSet.of() | DataSet.id() | Session.batch() | BatchOperationBuilder.upsert() | BatchKeyOperationBuilder.bin() | BatchBinBuilder.set_to() | BatchBinBuilder.add() | BatchOperationBuilder.execute()

Batch delete

Delete multiple records:

Java

DataSet users = DataSet.of("test", "users");

RecordStream result = session.delete(users.ids("user-1", "user-2", "user-3"))
    .execute();

// Check which deletes succeeded
int i = 0;
while (result.hasNext()) {
    RecordResult rr = result.next();
    System.out.println("Record " + i++ + " deleted: " + rr.asBoolean());
}
result.close();

📖 API reference: DataSet.ids(...) | Session.delete(List) | Session.delete(Key) | ChainableQueryBuilder.execute() | RecordStream.hasNext() | RecordStream.next() | RecordStream.close()

Python

users = DataSet.of("test", "users")

stream = await session.delete(users.ids("user-1", "user-2", "user-3")).execute()

# Check which deletes succeeded
async for rr in stream:
    print(f"Record {rr.index} deleted: {rr.as_bool()}")
stream.close()

📖 API reference: DataSet.of() | DataSet.ids() | Session.delete() | WriteSegmentBuilder.execute() | RecordResult.index | RecordResult.as_bool() | RecordStream.close()

Mixed batch operations

Combine different operation types in one call. Note that the operations on each key are performed asynchronously across the nodes in the cluster, and hence the returned results may not be in the same order as the commands. Each returned item exposes a key and an index — key() and index() in Java, the key and index attributes in Python (row.key, row.index). key is the unique key of the record and index is the zero-based index of the command in the original list.

Java

DataSet users = DataSet.of("test", "users");

RecordStream stream = session
    .query(users.ids("user-1", "user-2"))
    .upsert(users.id("user-3"))
        .bin("status").setTo("active")
    .delete(users.id("user-4"))
    .execute();

stream.forEach(result -> {
    switch (result.index()) {
        case 0 -> handleUser1(result);
        case 1 -> handleUser2(result);
        case 2 -> System.out.println("Upsert: " + (result.isOk() ? "ok" : result.message()));
        case 3 -> System.out.println("Delete: " + (result.isOk() ? "ok" : result.message()));
    }
});

📖 API reference: DataSet.ids(...) | DataSet.id(...) | ChainableQueryBuilder.bin(...) | ChainableQueryBuilder.execute() | RecordStream.forEach(...) | RecordResult.isOk()

Python

users = DataSet.of("test", "users")

stream = await (
    session.query(users.ids("user-1", "user-2"))
    .upsert(users.id("user-3")).bin("status").set_to("active")
    .delete(users.id("user-4"))
    .execute()
)
async for row in stream:
    match row.index:
        case 0: handle_user1(row)
        case 1: handle_user2(row)
        case 2: print(f"Upsert: {'ok' if row.is_ok else row.result_code}")
        case 3: print(f"Delete: {'ok' if row.is_ok else row.result_code}")
stream.close()

📖 API reference: DataSet.of() | DataSet.id() | DataSet.ids() | Session.query() | RecordResult.index | RecordResult.is_ok | RecordStream.close()

Batch with selected bins

Read only specific bins in batch:

Java

DataSet users = DataSet.of("test", "users");

RecordStream result = session.query(users.ids("user-1", "user-2", "user-3"))
    .bins("name", "email")
    .execute();

📖 API reference: DataSet.ids(...) | Session.query(List) | OperationObjectBuilder.bins(...) | ChainableQueryBuilder.execute()

Python

users = DataSet.of("test", "users")

stream = await (
    session.query(users.ids("user-1", "user-2", "user-3"))
    .bins(["name", "email"])
    .execute()
)

📖 API reference: DataSet.of() | DataSet.ids() | Session.query() | QueryBuilder.bins() | QueryBuilder.execute()

Handle partial failures

Some operations in a batch may fail while others succeed:

Java

DataSet users = DataSet.of("test", "users");

RecordStream result = session
    .update(users.ids("user-1", "nonexistent", "user-3"))
    .bin("lastSeen").setTo(System.currentTimeMillis())
    .execute(ErrorStrategy.IN_STREAM);  // Place errors in the RecordStream
result.forEach(rr -> {
    String keyId = rr.key().userKey.toString();
    if (!rr.isOk()) {
        System.out.println(keyId + " failed: " + rr.message());
    } else {
        System.out.println(keyId + ": updated");
    }
});

📖 API reference: DataSet.ids(...) | ChainableQueryBuilder.bin(...) | ChainableQueryBuilder.execute() | RecordStream.forEach(...) | RecordResult.isOk() | ErrorStrategy | ErrorStrategy.IN_STREAM

Python

# Additional imports for this example:
import time
from aerospike_sdk import ErrorStrategy

users = DataSet.of("test", "users")

stream = await (
    session.batch()
    .update(users.id("user-1")).bin("lastSeen").set_to(int(time.time() * 1000))
    .update(users.id("nonexistent")).bin("lastSeen").set_to(int(time.time() * 1000))
    .update(users.id("user-3")).bin("lastSeen").set_to(int(time.time() * 1000))
    .execute(on_error=ErrorStrategy.IN_STREAM)  # Place errors in the RecordStream
)
async for row in stream:
    key_id = row.key.value
    if not row.is_ok:
        print(f"{key_id} failed: {row.result_code}")
    else:
        print(f"{key_id}: updated")
stream.close()

📖 API reference: DataSet.of() | DataSet.id() | Session.batch() | BatchOperationBuilder.update() | BatchKeyOperationBuilder.bin() | BatchBinBuilder.set_to() | BatchOperationBuilder.execute() | ErrorStrategy.IN_STREAM | RecordResult.key | RecordResult.is_ok | RecordResult.result_code | RecordStream.close()

Dynamic batch building

Build batches programmatically:

Java

// Additional imports for this example:
import com.aerospike.client.sdk.IdValuesRowBuilder;

DataSet users = DataSet.of("test", "users");

IdValuesRowBuilder rows = session.upsert(users)
    .bins("name", "score")
    .id("user-1").values("User 1", 10);
for (int id = 2; id <= 50; id++) {
    rows.id("user-" + id).values("User " + id, id * 10);
}
rows.execute();

📖 API reference: DataSet.id(...) | Session.upsert(DataSet) | OperationObjectBuilder.bins(...) | IdValuesBuilder.id(...) | IdValuesRowBuilder.values(...) | ChainableQueryBuilder.execute()

Python

users = DataSet.of("test", "users")

batch = session.batch()
for i in range(1, 51):
    batch.upsert(users.id(f"user-{i}")).put({"name": f"User {i}", "score": i * 10})
await batch.execute()

📖 API reference: DataSet.of() | DataSet.id() | Session.batch() | BatchOperationBuilder.upsert() | BatchKeyOperationBuilder.put() | BatchOperationBuilder.execute()

Complete example

This example is self-contained—it lists every import needed to run standalone.

Java

import com.aerospike.client.sdk.Cluster;
import com.aerospike.client.sdk.ClusterDefinition;
import com.aerospike.client.sdk.DataSet;
import com.aerospike.client.sdk.Record;
import com.aerospike.client.sdk.RecordResult;
import com.aerospike.client.sdk.RecordStream;
import com.aerospike.client.sdk.Session;
import com.aerospike.client.sdk.policy.Behavior;

public class BatchOperationsExample {
    public static void main(String[] args) {
        try (Cluster cluster = new ClusterDefinition("localhost", 3000).connect()) {
            Session session = cluster.createSession(Behavior.DEFAULT);
            DataSet users = DataSet.of("test", "users");
            String key1 = "batch-example-1";
            String key2 = "batch-example-2";
            String key3 = "batch-example-3";

            // Cleanup so the example is repeatable.
            session.delete(users.ids(key1, key2, key3)).execute().close();

            // Batch insert
            session.insert(users)
                .bins("name", "age")
                .id(key1).values("Alice", 28)
                .id(key2).values("Bob", 35)
                .id(key3).values("Carol", 22)
                .execute();
            System.out.println("Batch insert complete");

            // Batch read
            RecordStream readStream = session.query(users.ids(key1, key2, key3))
                .execute()
                .forEach(result -> {
                    Record record = result.recordOrThrow();
                    System.out.println("  - " + record.getString("name"));
                });

            // Batch update
            session
                .upsert(users.id(key1), users.id(key2))
                    .bin("status").setTo("active")
                .upsert(users.id(key3))
                    .bin("status").setTo("inactive")
                .execute().close();
            System.out.println("\nBatch update complete");

            // Batch delete
            session.delete(users.ids(key1, key2, key3))
                .execute().close();
            System.out.println("Batch delete complete");
        }
    }
}

📖 API reference: ClusterDefinition(String,int) | ClusterDefinition.connect() | Cluster.createSession(Behavior) | Cluster.close() | DataSet.of(...) | DataSet.ids(...) | Session.insert(DataSet) | Session.delete(List) | Session.delete(Key) | Session.query(List) | OperationObjectBuilder.bins(...) | IdValuesBuilder.id(...) | IdValuesRowBuilder.values(...) | ChainableQueryBuilder.bin(...) | ChainableQueryBuilder.execute() | ChainableNoBinsBuilder.execute() | RecordStream.forEach(...) | RecordStream.close() | RecordResult.recordOrThrow() | Record.getString(...)

Python

import asyncio
from aerospike_sdk import Behavior, ClusterDefinition, DataSet


async def main():
    async with await ClusterDefinition("localhost", 3000).connect() as cluster:
        session = cluster.create_session(Behavior.DEFAULT)
        users = DataSet.of("test", "users")
        key1 = users.id("batch-example-1")
        key2 = users.id("batch-example-2")
        key3 = users.id("batch-example-3")

        # Cleanup so the example is repeatable.
        stream = await session.delete(key1, key2, key3).execute()
        stream.close()

        # Batch insert
        await (
            session.batch()
            .insert(key1).put({"name": "Alice", "age": 28})
            .insert(key2).put({"name": "Bob",   "age": 35})
            .insert(key3).put({"name": "Carol", "age": 22})
            .execute()
        )
        print("Batch insert complete")

        # Batch read
        stream = await session.query(key1, key2, key3).execute()
        print("\nBatch read complete:")
        async for row in stream:
            if row.is_ok:
                print(f"  - {row.record_or_raise().bins['name']}")
        stream.close()

        # Batch update
        await (
            session.batch()
            .upsert(key1).bin("status").set_to("active")
            .upsert(key2).bin("status").set_to("active")
            .upsert(key3).bin("status").set_to("inactive")
            .execute()
        )
        print("\nBatch update complete")

        # Batch delete
        stream = await session.delete(key1, key2, key3).execute()
        stream.close()
        print("Batch delete complete")


if __name__ == "__main__":
    asyncio.run(main())

📖 API reference: ClusterDefinition | ClusterDefinition.connect() | Cluster.create_session() | DataSet.of() | DataSet.id() | Behavior.DEFAULT | Session.query() | Session.delete() | Session.batch() | BatchOperationBuilder.insert() | BatchOperationBuilder.upsert() | BatchKeyOperationBuilder.put() | BatchKeyOperationBuilder.bin() | BatchBinBuilder.set_to() | BatchOperationBuilder.execute() | QueryBuilder.execute() | WriteSegmentBuilder.execute() | RecordResult.is_ok | RecordResult.record_or_raise() | RecordStream.close()

API reference summary

Java	Python	Description
session.query(dataSet.ids(...))	await session.query(data_set.ids(...)).execute()	Batch-read multiple record IDs
session.insert(dataSet) + repeated .id().values()	session.batch().insert(key).put({...}) (one per key)	Batch insert with one request
session.upsert(dataSet) + repeated .id().values()	session.batch().upsert(key).put({...}) (one per key)	Batch upsert with one request
session.delete(dataSet.ids(...))	await session.batch().delete(key)…execute() (one per key)	Batch delete multiple IDs
.bins(...) (varargs)	.bins([...]) (list)	Select projected bins for batch reads
RecordStream / forEach	RecordStream / async for	Iterate per-record results

Next steps

Async Operations

Non-blocking operations for high throughput.

Async Operations →

Query Records

Find records with DSL queries.

Query Records →

Behaviors

Configure batch timeouts and retries.

Behaviors →