Change your application: migrate from Aerospike to KiviDB

Change your application. Migrate to KiviDB: Move from Redis, Valkey, ElastiCache, Redis Cloud, Aiven, Aerospike, Hazelcast and more: getting the data out, loa

Connecting

Python, before and after:

# Before: Aerospike
import aerospike
client = aerospike.client({"hosts": [("aerospike.internal", 3000)]}).connect()

# After: KiviDB, self-hosted
import redis
r = redis.Redis(host="kividb.internal", port=6380, password=os.environ["KIVIDB_PASSWORD"],
                decode_responses=True)

# After: KiviDB Cloud, with TLS
r = redis.Redis.from_url(f"rediss://:{os.environ['KIVIDB_PASSWORD']}@your-name.cloud.kividb.io:6380",
                         ssl_ca_certs="kividb-ca.pem", decode_responses=True)

Java (Jedis 5):

// Before: Aerospike
AerospikeClient client = new AerospikeClient("aerospike.internal", 3000);

// After: KiviDB
JedisPooled jedis = new JedisPooled(new HostAndPort("kividb.internal", 6380),
    DefaultJedisClientConfig.builder().password(System.getenv("KIVIDB_PASSWORD")).build());

// After: KiviDB Cloud, with TLS. rediss:// turns TLS on; trust the CA with a trust store:
//   keytool -importcert -noprompt -alias kividb -file kividb-ca.pem -keystore kividb.jks -storepass changeit
//   java -Djavax.net.ssl.trustStore=kividb.jks -Djavax.net.ssl.trustStorePassword=changeit ...
JedisPooled jedis = new JedisPooled("rediss://:" + System.getenv("KIVIDB_PASSWORD")
    + "@your-name.cloud.kividb.io:6380");

Go (go-redis v9):

// Before: Aerospike
client, err := as.NewClient("aerospike.internal", 3000)

// After: KiviDB
rdb := redis.NewClient(&redis.Options{
	Addr: "kividb.internal:6380", Password: os.Getenv("KIVIDB_PASSWORD"),
})

// After: KiviDB Cloud, with TLS
pem, _ := os.ReadFile("kividb-ca.pem")
pool := x509.NewCertPool()
pool.AppendCertsFromPEM(pem)
rdb := redis.NewClient(&redis.Options{
	Addr: "your-name.cloud.kividb.io:6380", Password: os.Getenv("KIVIDB_PASSWORD"),
	TLSConfig: &tls.Config{RootCAs: pool},
})

Connect to KiviDB Cloud by hostname, not IP address: the certificate names the host.

Operations at a glance

Aerospike	KiviDB, JSON	KiviDB, hash
`put(key, bins, ttl)`	`JSON.SET k $ doc` + `EXPIRE k ttl` in `MULTI`	`HSET k f v ...` + `EXPIRE` in `MULTI`
`get(key)`	`JSON.GET k`	`HGETALL k`
`select(key, bins)` / `get(key, bins...)`	`JSON.GET k $.a $.b`	`HMGET k a b`
`operate(add, get)`	`JSON.NUMINCRBY k $.n 1` (returns the new value)	`HINCRBY k n 1`
List append	`JSON.ARRAPPEND k $.list '"x"'`	Read, change and write the field
Map put	`JSON.SET k $.map.key '"v"'`	Read, change and write the field
Batch get	`JSON.MGET k1 k2 $`, or a pipeline	A pipeline of `HGETALL`
`exists(key)`	`EXISTS k`	`EXISTS k`
`remove(key)`	`DEL k`	`DEL k`
`touch(key, ttl)`	`EXPIRE k ttl`	`EXPIRE k ttl`
TTL in the record metadata	`TTL k` (`-1`: never expires)	`TTL k`
Generation check	`WATCH` + version field + `MULTI`	Same
Query by secondary index	`FT.SEARCH idx "@age:[30 30]"`	Same
Geo query	`GEOSEARCH geo:app:users:loc FROMLONLAT lon lat BYRADIUS 1 km`	Same
Truncate a set	`SCAN` with `MATCH app:users:*` and `DEL` (or `UNLINK`)	Same

Python, side by side

Before:

import aerospike
from aerospike import predicates as p
from aerospike_helpers.operations import operations as op
from aerospike_helpers.operations import list_operations as lop
from aerospike_helpers.operations import map_operations as mop

client = aerospike.client({"hosts": [("aerospike.internal", 3000)]}).connect()
key = ("app", "users", "u42")

client.put(key, {"name": "Ana", "age": 31, "tags": ["new"], "prefs": {"theme": "dark"}},
           meta={"ttl": 3600}, policy={"key": aerospike.POLICY_KEY_SEND})
_, meta, bins = client.get(key)
_, _, some = client.select(key, ["name", "age"])
_, _, out = client.operate(key, [op.increment("age", 1), op.read("age")])
client.operate(key, [lop.list_append("tags", "vip"),
                     mop.map_put("prefs", "lang", "pt")])
res = client.batch_read([("app", "users", "u42"), ("app", "users", "u43")])
recs = [br.record for br in res.batch_records]
client.touch(key, 7200)

q = client.query("app", "users")
q.where(p.equals("age", 30))
matches = q.results()

client.remove(key)

After:

import redis
from redis.commands.search.query import Query

r = redis.Redis(host="kividb.internal", port=6380, password=PASSWORD, decode_responses=True)
key = "app:users:u42"

with r.pipeline() as pipe:            # MULTI/EXEC: both or neither
    pipe.json().set(key, "$", {"name": "Ana", "age": 31, "tags": ["new"], "prefs": {"theme": "dark"}})
    pipe.expire(key, 3600)
    pipe.execute()
doc = r.json().get(key)
some = r.json().get(key, "$.name", "$.age")     # {'$.name': ['Ana'], '$.age': [31]}
age = r.json().numincrby(key, "$.age", 1)       # [32]
with r.pipeline() as pipe:
    pipe.json().arrappend(key, "$.tags", "vip")
    pipe.json().set(key, "$.prefs.lang", "pt")
    pipe.execute()
docs = r.json().mget(["app:users:u42", "app:users:u43"], "$")   # [[{...}], None]
r.expire(key, 7200)

res = r.ft("idx:users").search(Query("@age:[30 30]").paging(0, 100))
matches = res.docs                    # res.total is the full count

r.delete(key)

With hashes instead of JSON, the same record is r.hset(key, mapping={"name": "Ana", "age": 31}), read with r.hgetall(key) or r.hmget(key, ["name", "age"]), and incremented with r.hincrby(key, "age", 1). Values come back as strings.

Java, side by side

Before:

AerospikeClient client = new AerospikeClient("aerospike.internal", 3000);
Key key = new Key("app", "users", "u42");

WritePolicy wp = new WritePolicy();
wp.sendKey = true;           // store the key so a scan can return it
wp.expiration = 3600;        // TTL in seconds
client.put(wp, key,
    new Bin("name", "Ana"), new Bin("age", 31),
    new Bin("tags", List.of("new")), new Bin("prefs", Map.of("theme", "dark")));

Record rec = client.get(null, key);
Record some = client.get(null, key, "name", "age");
Record inc = client.operate(null, key, Operation.add(new Bin("age", 1)), Operation.get("age"));
client.operate(null, key,
    ListOperation.append("tags", Value.get("vip")),
    MapOperation.put(MapPolicy.Default, "prefs", Value.get("lang"), Value.get("pt")));
Record[] batch = client.get(null, new Key[] {key, new Key("app", "users", "u43")});

WritePolicy touch = new WritePolicy();
touch.expiration = 7200;
client.touch(touch, key);

Statement stmt = new Statement();
stmt.setNamespace("app");
stmt.setSetName("users");
stmt.setFilter(Filter.equal("age", 30));
try (RecordSet rs = client.query(null, stmt)) {
  while (rs.next()) { /* rs.getRecord() */ }
}
client.delete(null, key);

After (import com.aerospike.client.Record explicitly if both clients are on the class path in one file; java.lang.Record has the same name):

JedisPooled jedis = new JedisPooled(new HostAndPort("kividb.internal", 6380),
    DefaultJedisClientConfig.builder().password(PASSWORD).build());
String k = "app:users:u42";

try (AbstractTransaction tx = jedis.multi()) {          // MULTI/EXEC
  tx.jsonSet(k, Path2.ROOT_PATH, new JSONObject()
      .put("name", "Ana").put("age", 31)
      .put("tags", new JSONArray().put("new"))
      .put("prefs", new JSONObject().put("theme", "dark")));
  tx.expire(k, 3600);
  tx.exec();
}
Object doc = jedis.jsonGet(k);                                  // the whole document
Object name = jedis.jsonGet(k, Path2.of("$.name"));            // ["Ana"]
Object age = jedis.jsonNumIncrBy(k, Path2.of("$.age"), 1);     // [32]
jedis.jsonArrAppend(k, Path2.of("$.tags"), "\"vip\"");
jedis.jsonSet(k, Path2.of("$.prefs.lang"), "\"pt\"");
List<JSONArray> docs = jedis.jsonMGet(Path2.ROOT_PATH, k, "app:users:u43");   // null for a missing key
jedis.expire(k, 7200);

SearchResult res = jedis.ftSearch("idx:users", "@age:[30 30]",
    FTSearchParams.searchParams().limit(0, 100));
jedis.del(k);

JSONObject and JSONArray are from org.json, which Jedis depends on; Path2 is redis.clients.jedis.json.Path2.

Go, side by side

Before:

client, err := as.NewClient("aerospike.internal", 3000)
key, _ := as.NewKey("app", "users", "u42")

wp := as.NewWritePolicy(0, 3600) // generation 0 (ignored), TTL 3600 s
wp.SendKey = true
err = client.Put(wp, key, as.BinMap{
	"name": "Ana", "age": 31,
	"tags": []any{"new"}, "prefs": map[string]any{"theme": "dark"},
})
rec, err := client.Get(nil, key)
some, err := client.Get(nil, key, "name", "age")
inc, err := client.Operate(nil, key, as.AddOp(as.NewBin("age", 1)), as.GetBinOp("age"))
_, err = client.Operate(nil, key,
	as.ListAppendOp("tags", "vip"),
	as.MapPutOp(as.DefaultMapPolicy(), "prefs", "lang", "pt"))
k2, _ := as.NewKey("app", "users", "u43")
recs, err := client.BatchGet(nil, []*as.Key{key, k2}) // nil entry for a missing key
err = client.Touch(as.NewWritePolicy(0, 7200), key)

stmt := as.NewStatement("app", "users")
err = stmt.SetFilter(as.NewEqualFilter("age", 30))
rs, err := client.Query(nil, stmt)
for res := range rs.Results() { /* res.Record, res.Err */ }
_, err = client.Delete(nil, key)

After:

ctx := context.Background()
rdb := redis.NewClient(&redis.Options{
	Addr: "kividb.internal:6380", Password: os.Getenv("KIVIDB_PASSWORD"),
	Protocol: 2, // FT.SEARCH replies parse in RESP2 without UnstableResp3
})
k := "app:users:u42"

_, err := rdb.TxPipelined(ctx, func(p redis.Pipeliner) error {
	p.JSONSet(ctx, k, "$", map[string]any{
		"name": "Ana", "age": 31, "tags": []any{"new"}, "prefs": map[string]any{"theme": "dark"},
	})
	p.Expire(ctx, k, time.Hour)
	return nil
})
doc, err := rdb.JSONGet(ctx, k).Result()                     // the document as a JSON string
some, err := rdb.JSONGet(ctx, k, "$.name", "$.age").Result() // {"$.name":["Ana"],"$.age":[31]}
age, err := rdb.JSONNumIncrBy(ctx, k, "$.age", 1).Result()   // "[32]"
_, err = rdb.TxPipelined(ctx, func(p redis.Pipeliner) error {
	p.JSONArrAppend(ctx, k, "$.tags", `"vip"`)
	p.JSONSet(ctx, k, "$.prefs.lang", `"pt"`)
	return nil
})
docs, err := rdb.JSONMGet(ctx, "$", k, "app:users:u43").Result() // nil for a missing key
rdb.Expire(ctx, k, 2*time.Hour)

res, err := rdb.FTSearchWithArgs(ctx, "idx:users", "@age:[30 30]",
	&redis.FTSearchOptions{LimitOffset: 0, Limit: 100}).Result()
rdb.Del(ctx, k)

Generation checks in code

Aerospike, Python:

_, meta, bins = client.get(key)
client.put(key, {"age": bins["age"] + 1},
           policy={"gen": aerospike.POLICY_GEN_EQ},
           meta={"gen": meta["gen"], "ttl": aerospike.TTL_DONT_UPDATE})

KiviDB, with a version field _v (written by --gen-field _v, and increased by every write from now on):

import json
import redis


def put_if_version(r, key, expected, path, value):
    with r.pipeline(transaction=True) as pipe:
        try:
            pipe.watch(key)
            current = pipe.json().get(key, "$._v")   # runs now, under WATCH
            if isinstance(current, str):             # RedisCluster gives the raw reply here
                current = json.loads(current)
            if current != [expected]:
                pipe.unwatch()
                return False                 # like a generation error
            pipe.multi()
            pipe.json().set(key, path, value)
            pipe.json().numincrby(key, "$._v", 1)
            pipe.execute()                   # raises WatchError if key changed since WATCH
            return True
        except redis.WatchError:
            return False

Java: WATCH, the read and MULTI/EXEC must use one connection. With JedisPooled, take one from the pool:

static boolean putIfVersion(JedisPooled jedis, String key, long expected, String path, Object value) {
  try (Connection conn = jedis.getPool().getResource()) {
    UnifiedJedis on = new UnifiedJedis(conn);
    Transaction tx = new Transaction(conn, false);
    tx.watch(key);
    JSONArray v = (JSONArray) on.jsonGet(key, Path2.of("$._v"));
    if (v == null || v.isEmpty() || v.getLong(0) != expected) {
      tx.unwatch();
      return false;
    }
    tx.multi();
    tx.jsonSet(key, Path2.of(path), value);
    tx.jsonNumIncrBy(key, Path2.of("$._v"), 1);
    return tx.exec() != null;     // null: the key changed after WATCH
  }
}

Go: go-redis's Watch runs the function on one connection and returns redis.TxFailedErr if the key changed:

err := rdb.Watch(ctx, func(tx *redis.Tx) error {
	v, err := tx.JSONGet(ctx, k, "$._v").Result()
	if err != nil {
		return err
	}
	if v != fmt.Sprintf("[%d]", expected) {
		return errors.New("version changed")
	}
	_, err = tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
		p.JSONSet(ctx, k, "$.age", 33)
		p.JSONNumIncrBy(ctx, k, "$._v", 1)
		return nil
	})
	return err
}, k)

For a read-modify-write inside one function (no version carried between requests), drop the version comparison and keep WATCH: EXEC fails if anything else wrote the key in between, and you retry.

Secondary-index and geo queries

Aerospike	KiviDB
`Filter.equal("age", 30)`	`FT.SEARCH idx:users "@age:[30 30]"`
`Filter.range("age", 30, 40)`	`FT.SEARCH idx:users "@age:[30 40]"`
`Filter.equal("country", "IN")`	`FT.SEARCH idx:users "@country:{IN}"`
Filter plus an expression on another bin	`FT.SEARCH idx:users "@country:{IN} @age:[30 40]"`
`Filter.contains("tags", LIST, "vip")`	`FT.SEARCH idx:users "@tags:{vip}"` (JSON, `$.tags[*]`)
Sorted results	`... SORTBY age ASC` (the field needs `SORTABLE`)
Count only	`... LIMIT 0 0` (the reply's total)
Aggregation (stream UDF)	`FT.AGGREGATE idx:users "*" GROUPBY 1 @country REDUCE COUNT 0 AS n`
`Filter.geoWithinRadius("loc", lon, lat, meters)`	`GEOSEARCH geo:app:users:loc FROMLONLAT lon lat BYRADIUS 1 km ASC WITHDIST`

GEOSEARCH returns key names; read the records with JSON.MGET or a pipeline. Keep the geo set current in the application: GEOADD when a location changes, ZREM when the record is deleted. Geo regions (polygons) and "contains" queries have no equivalent; filter in the application.

Search queries a word or a whole tag, not a substring of a tag value, and has no stemming. The command reference has the full query syntax.

UDFs, expressions and scans

Record UDFs (Lua that changes one record): rewrite as application code in a WATCH/MULTI block, or as one JSON command when the change is a path operation. KiviDB's Lua (EVAL) is Redis's API, not Aerospike's, and needs the lua or full build (a switch on KiviDB Cloud, set when the database is created).
Stream UDFs and aggregations: FT.AGGREGATE over a search index, or the application.
Expressions used as filters: express them as search queries on indexed fields. Expressions that compute values on the server have no equivalent; do it in the application.
Scans of a whole set: SCAN 0 MATCH app:users:* COUNT 1000, then read the keys. A search index over the set is usually better.