Load into KiviDB: migrate from Aerospike to KiviDB

The loader

as2kivi.py reads one namespace with the Aerospike Python client and writes it with redis-py, in pipelines of 500 records, eight partitions at a time.

• Key names: <namespace>:<set>:<user key>, or <namespace>:<set>:#<digest in hex> for a record stored without its key (--no-key skip leaves those out).
• --mode hash (default) or --mode json, with the conversions in Bin types.
• TTLs carried over as an absolute expiry.
• --geo adds every GeoJSON point to a geo set geo:<namespace>:<set>:<bin>, with the record's key name as the member.
• --gen-field _v writes the generation into a field.
• Resumable: each finished partition is appended to a checkpoint file (as2kivi.done), and a second run skips those. Writing a record deletes the key first, so running a partition twice is safe.
• --since <unix time> copies only the records written after that time.
• --prune removes keys whose record no longer exists.
• --cluster for a sharded KiviDB database; rediss:// and --ca for TLS.

Save it as as2kivi.py:

#!/usr/bin/env python3
"""Copy an Aerospike namespace into KiviDB.

Each record becomes one KiviDB key, named <namespace>:<set>:<user key>. The
bins become hash fields (--mode hash) or a JSON document (--mode json). The
remaining TTL is carried over as an absolute expiry time.

The scan runs partition by partition (Aerospike has 4096), and each finished
partition is written to the checkpoint file, so a stopped run picks up where
it left off. Writing a record again replaces it, so re-running is safe.

    pip install aerospike redis
    python as2kivi.py --aerospike 10.0.0.5:3000 --namespace app \
        --kividb redis://:PASSWORD@kividb.internal:6380 --mode hash
"""
import argparse
import base64
import json
import os
import re
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

import aerospike
import redis
from aerospike_helpers import expressions as exp
from aerospike_helpers import HyperLogLog

N_PARTITIONS = 4096
NEVER_EXPIRES = 0xFFFFFFFF  # what clients report as the TTL of a record with no expiry


def as_json(value):
    """JSON for list and map bins. Bytes inside become {"$b64": "..."}."""
    def default(o):
        if isinstance(o, (bytes, bytearray)):
            return {"$b64": base64.b64encode(bytes(o)).decode()}
        if isinstance(o, aerospike.GeoJSON):
            return json.loads(o.dumps())
        raise TypeError(f"cannot convert {type(o).__name__}")
    # Map keys that are not strings (integers, bytes) become strings in JSON.
    return json.dumps(value, default=default, separators=(",", ":"))


def key_part(user_key):
    if isinstance(user_key, (bytes, bytearray)):
        return bytes(user_key)
    return str(user_key).encode()


class Stats:
    def __init__(self):
        self.lock = threading.Lock()
        self.records = self.no_key = self.hll_bins = self.geo_points = self.empty = 0
        self.pruned = 0

    def add(self, **kw):
        with self.lock:
            for k, v in kw.items():
                setattr(self, k, getattr(self, k) + v)


def convert_hash(bins, stats):
    """Bins -> hash fields. Returns (fields, geo points)."""
    fields, geo = {}, []
    for name, v in bins.items():
        if isinstance(v, HyperLogLog):  # check before bytes: it is a bytes subclass
            stats.add(hll_bins=1)
            continue
        if isinstance(v, aerospike.GeoJSON):
            s = v.dumps()
            fields[name] = s
            g = json.loads(s)
            if g.get("type") == "Point":
                geo.append((name, g["coordinates"][0], g["coordinates"][1]))
        elif isinstance(v, bool):  # before int: bool is an int subclass
            fields[name] = "1" if v else "0"
        elif isinstance(v, int):
            fields[name] = str(v)
        elif isinstance(v, float):
            fields[name] = repr(v)
        elif isinstance(v, str):
            fields[name] = v
        elif isinstance(v, (bytes, bytearray)):
            fields[name] = bytes(v)
        elif isinstance(v, (list, dict)):
            fields[name] = as_json(v)
        else:
            raise TypeError(f"bin {name}: unsupported type {type(v).__name__}")
    return fields, geo


def convert_json(bins, stats):
    doc, geo = {}, []
    for name, v in bins.items():
        if isinstance(v, HyperLogLog):
            stats.add(hll_bins=1)
            continue
        if isinstance(v, aerospike.GeoJSON):
            g = json.loads(v.dumps())
            doc[name] = g
            if g.get("type") == "Point":
                geo.append((name, g["coordinates"][0], g["coordinates"][1]))
        else:
            doc[name] = json.loads(as_json(v))
    return doc, geo


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--aerospike", required=True, help="host:port of any node")
    ap.add_argument("--as-user"); ap.add_argument("--as-password")
    ap.add_argument("--namespace", required=True)
    ap.add_argument("--set", action="append", dest="sets",
                    help="copy only these sets (repeat the flag); default: every set")
    ap.add_argument("--kividb", required=True,
                    help="redis://[:password@]host:port, or rediss:// for TLS")
    ap.add_argument("--ca", help="CA certificate file for rediss://")
    ap.add_argument("--cluster", action="store_true", help="the target is a cluster (sharded) database")
    ap.add_argument("--mode", choices=["hash", "json"], default="hash")
    ap.add_argument("--prefix", default="", help="put before every key name")
    ap.add_argument("--no-key", choices=["digest", "skip"], default="digest",
                    help="records whose key was not stored: name them by digest, or skip them")
    ap.add_argument("--gen-field", help="also write the record generation to this field (e.g. _v)")
    ap.add_argument("--geo", action="store_true",
                    help="also GEOADD GeoJSON points to geo:<ns>:<set>:<bin>")
    ap.add_argument("--workers", type=int, default=8, help="partitions scanned at once")
    ap.add_argument("--batch", type=int, default=500, help="records per pipeline round trip")
    ap.add_argument("--checkpoint", help="file listing finished partitions "
                    "(default as2kivi.done, or as2kivi-since-<time>.done with --since)")
    ap.add_argument("--since", type=float,
                    help="copy only records written after this Unix time (a catch-up pass)")
    ap.add_argument("--prune", action="store_true",
                    help="instead of copying, delete KiviDB keys whose record no longer exists")
    a = ap.parse_args()

    host, port = a.aerospike.rsplit(":", 1)
    as_cfg = {"hosts": [(host, int(port))], "policies": {"query": {"total_timeout": 0, "socket_timeout": 30000}}}
    if a.as_user:
        as_cfg["user"], as_cfg["password"] = a.as_user, a.as_password
    asc = aerospike.client(as_cfg).connect()

    kw = {"ssl_ca_certs": a.ca} if a.ca else {}
    if a.cluster:
        kv = redis.RedisCluster.from_url(a.kividb, **kw)
    else:
        kv = redis.Redis.from_url(a.kividb, max_connections=a.workers * 2, **kw)
    kv.ping()

    if a.prune:
        prune(a, asc, kv)
        return

    if not a.checkpoint:
        a.checkpoint = f"as2kivi-since-{int(a.since)}.done" if a.since else "as2kivi.done"
    done = set()
    if os.path.exists(a.checkpoint):
        with open(a.checkpoint) as f:
            done = {int(line) for line in f if line.strip()}
    todo = [p for p in range(N_PARTITIONS) if p not in done]
    print(f"{len(done)} partitions already copied, {len(todo)} to go", file=sys.stderr)

    ckpt_lock = threading.Lock()
    ckpt = open(a.checkpoint, "a")
    stats = Stats()
    ns = a.namespace
    wanted = set(a.sets) if a.sets else None
    convert = convert_hash if a.mode == "hash" else convert_json

    def copy_partition(pid):
        pipe = kv.pipeline(transaction=False)
        pending = 0

        def write(record):
            nonlocal pending
            key, meta, bins = record[0], record[1], record[2]
            _, set_name, user_key, digest = key
            set_name = set_name or ""
            if wanted is not None and set_name not in wanted:
                return
            if user_key is None:
                stats.add(no_key=1)
                if a.no_key == "skip":
                    return
                kpart = b"#" + bytes(digest).hex().encode()
            else:
                kpart = key_part(user_key)
            name = f"{a.prefix}{ns}:{set_name}:".encode() + kpart
            body, geo = convert(bins, stats)
            if a.gen_field:
                body[a.gen_field] = meta["gen"] if a.mode == "json" else str(meta["gen"])
            if not body:
                stats.add(empty=1)  # every bin was a HyperLogLog
                return
            pipe.delete(name)
            if a.mode == "hash":
                pipe.hset(name, mapping=body)
            else:
                pipe.execute_command("JSON.SET", name, "$", json.dumps(body, separators=(",", ":")))
            ttl = meta["ttl"]
            if ttl != NEVER_EXPIRES:
                pipe.expireat(name, int(time.time()) + ttl)
            if a.geo:
                for bin_name, lon, lat in geo:
                    pipe.geoadd(f"{a.prefix}geo:{ns}:{set_name}:{bin_name}", (lon, lat, name))
                    stats.add(geo_points=1)
            stats.add(records=1)
            pending += 1
            if pending >= a.batch:
                pipe.execute()
                pending = 0

        policy = {"partition_filter": {"begin": pid, "count": 1}}
        if a.since:
            # last-update-time is in nanoseconds since the Unix epoch
            policy["expressions"] = exp.GT(exp.LastUpdateTime(), int(a.since * 1e9)).compile()
        q = asc.query(ns, None)  # no filter: every record in the partition
        q.foreach(lambda _pid, rec: write(rec), policy)
        if pending:
            pipe.execute()
        with ckpt_lock:
            ckpt.write(f"{pid}\n")
            ckpt.flush()

    started = time.time()
    with ThreadPoolExecutor(a.workers) as ex:
        futures = [ex.submit(copy_partition, p) for p in todo]
        for i, fut in enumerate(as_completed(futures), 1):
            fut.result()  # raise the first error; finished partitions stay checkpointed
            if i % 256 == 0:
                print(f"{i}/{len(todo)} partitions, {stats.records} records", file=sys.stderr)
    ckpt.close()
    print(f"copied {stats.records} records in {time.time() - started:.1f}s; "
          f"{stats.no_key} had no stored key ({a.no_key}); "
          f"{stats.hll_bins} HyperLogLog bins skipped ({stats.empty} records had nothing else); "
          f"{stats.geo_points} geo points indexed")


def candidates(ns, set_name, kpart):
    """The Aerospike keys a KiviDB key name could have come from."""
    s = set_name or None
    if re.fullmatch(rb"#[0-9a-f]{40}", kpart):
        return [(ns, s, None, bytearray(bytes.fromhex(kpart[1:].decode())))]
    keys = [(ns, s, bytearray(kpart))]
    try:
        text = kpart.decode()
        keys.append((ns, s, text))
        if re.fullmatch(r"-?[0-9]+", text) and -2**63 <= int(text) < 2**63:
            keys.append((ns, s, int(text)))
    except UnicodeDecodeError:
        pass
    return keys


def prune(a, asc, kv):
    """Delete KiviDB keys whose Aerospike record is gone (deleted during the copy)."""
    head = f"{a.prefix}{a.namespace}:".encode()
    stats = Stats()
    batch = []

    def flush():
        keys = [k for _, cands in batch for k in cands]
        found = set()
        for i, br in enumerate(asc.batch_read(keys, []).batch_records):
            if br.result == 0:
                found.add(i)
        i = 0
        for name, cands in batch:
            if not any(j in found for j in range(i, i + len(cands))):
                set_name = name[len(head):].split(b":", 1)[0].decode()
                kv.delete(name)
                if a.geo:
                    for g in kv.scan_iter(match=f"{a.prefix}geo:{a.namespace}:{set_name}:*"):
                        kv.zrem(g, name)
                stats.add(pruned=1)
            i += len(cands)
        batch.clear()

    for name in kv.scan_iter(match=head + b"*", count=1000):
        set_name, kpart = name[len(head):].split(b":", 1)
        if a.sets and set_name.decode() not in a.sets:
            continue
        batch.append((name, candidates(a.namespace, set_name.decode(), kpart)))
        if len(batch) >= 300:
            flush()
    if batch:
        flush()
    print(f"pruned {stats.pruned} keys whose record no longer exists")


if __name__ == "__main__":
    main()

Install the two clients in a virtual environment (Python 3.9 or later):

python3 -m venv as2kivi-env
as2kivi-env/bin/pip install aerospike redis

With redis-py 8.1 on Python 3.10, also install typing_extensions; that release imports it without declaring it, and the search commands fail without it.

Run the loader on a machine close to both servers: it is bound by round trips. --workers sets how many partitions are read at once; raise it until the Aerospike or KiviDB side is busy.

On Enterprise Edition with security on, pass --as-user and --as-password; the user needs read access to the namespace. Community Edition has no users.

Self-hosted KiviDB

python as2kivi.py --aerospike 10.0.0.5:3000 --namespace app \
  --kividb redis://:"$KIVIDB_PASSWORD"@kividb.internal:6380 --mode json --geo

The output ends with a summary, for example:

copied 10201 records in 8.8s; 200 had no stored key (digest); 5000 HyperLogLog bins skipped (0 records had nothing else); 0 geo points indexed

If the run stops (a network error, a deploy, Ctrl-C), run the same command again. It reports how many partitions were already copied and continues. Delete as2kivi.done to start over.

To put a namespace in its own database instead of behind a prefix, add the database number to the URL: redis://:PASSWORD@kividb.internal:6380/2.

KiviDB Cloud

Connect over TLS with the endpoint, password and CA certificate from the database's overview (see Connect to your instance):

python as2kivi.py --aerospike 10.0.0.5:3000 --namespace app \
  --kividb rediss://:"$KIVIDB_PASSWORD"@your-name.cloud.kividb.io:6380 \
  --ca kividb-ca.pem --mode json --geo

The machine running the loader must be on the database's allow-list.

For a sharded database, add --cluster and connect to the cluster endpoint. Keys spread over the shards by name. The MULTI/EXEC and WATCH code in this guide touches one key at a time, so it works on a sharded database too; with redis-py's RedisCluster, open the pipeline with pipeline(transaction=True).

Loading offline instead. For a large namespace, or when the Cloud database is far from Aerospike, load into a self-hosted KiviDB next to Aerospike, take a snapshot, and import it:

redis-cli -p 6380 -a "$KIVIDB_PASSWORD" BGSAVE   # writes dump.kdb in the server's --dir

In the console, Snapshots → Import a snapshot, upload dump.kdb, then create a database from it or restore it into an existing one. The file can't be larger than the database's data size. Records copied this way keep their expiry times. Importing does not create a sharded database; load a sharded one directly with --cluster.

Rebuild the secondary indexes

Create a search index for each set you query. It indexes the keys already there and every key written afterwards. Create it after the load (or before; both work).

For hashes:

FT.CREATE idx:users ON HASH PREFIX 1 app:users: SCHEMA
  name TEXT
  age NUMERIC SORTABLE
  country TAG
  score NUMERIC
  active TAG

For JSON documents, name each path:

FT.CREATE idx:users ON JSON PREFIX 1 app:users: SCHEMA
  $.name AS name TEXT
  $.age AS age NUMERIC SORTABLE
  $.country AS country TAG
  $.tags[*] AS tags TAG
  $.prefs.theme AS theme TAG
  $.active AS active TAG

(Each FT.CREATE is one line; it is wrapped here to read.)

Aerospike index	KiviDB field
Integer index on a bin	NUMERIC (add SORTABLE to sort by it)
String index, equality	TAG
String bin you search by words	TEXT
Blob index	TAG on the hash field
List index (list collection type)	JSON: TAG or NUMERIC on $.bin[*]. A TAG field matches string elements only.
Map keys / map values index	JSON: a field per path you query, such as $.prefs.theme
Geo2DSphere index	Not a search field: use the --geo sets and GEOSEARCH
Expression index (8.1+)	Store the computed value as its own field, and index that

Booleans: in hash mode they are 1/0, so query @active:{1}; in JSON mode @active:{true}.

A Java loader

If your team would rather run Java, this does the same copy in hash mode, one partition at a time, with the same key names, conversions, TTLs and checkpoint file. It uses the Aerospike Java client (aerospike-client-jdk21 10.x), Jedis 5 and Gson, and Java 21. The verify script below checks its output too.

import com.aerospike.client.AerospikeClient;
import com.aerospike.client.Key;
import com.aerospike.client.Record;
import com.aerospike.client.Value;
import com.aerospike.client.query.PartitionFilter;
import com.aerospike.client.query.RecordSet;
import com.aerospike.client.query.Statement;
import com.google.gson.Gson;
import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import java.util.*;
import redis.clients.jedis.*;

// Copy one Aerospike namespace into KiviDB hashes: <ns>:<set>:<key>, one field
// per bin, TTL carried over. Partition by partition, with a checkpoint file.
// Usage: java AsToKivi <aerospike-host> <port> <namespace> <kividb-url> [checkpoint]
public class AsToKivi {
  static final Gson GSON = new Gson();

  static Object jsonSafe(Object v) {         // byte[] inside lists and maps -> {"$b64": ...}
    if (v instanceof byte[] b) return Map.of("$b64", Base64.getEncoder().encodeToString(b));
    if (v instanceof List<?> l) { List<Object> o = new ArrayList<>(); for (Object x : l) o.add(jsonSafe(x)); return o; }
    if (v instanceof Map<?, ?> m) { Map<String, Object> o = new LinkedHashMap<>(); m.forEach((k, x) -> o.put(String.valueOf(k), jsonSafe(x))); return o; }
    return v;
  }

  static byte[] field(Object v) {
    if (v instanceof byte[] b) return b;
    if (v instanceof Boolean b) return (b ? "1" : "0").getBytes();
    if (v instanceof Double d) return Double.toString(d).getBytes();
    if (v instanceof Value.GeoJSONValue g) return g.getObject().toString().getBytes(StandardCharsets.UTF_8);
    if (v instanceof List || v instanceof Map) return GSON.toJson(jsonSafe(v)).getBytes(StandardCharsets.UTF_8);
    return String.valueOf(v).getBytes(StandardCharsets.UTF_8);   // Long, String
  }

  public static void main(String[] a) throws Exception {
    String ns = a[2];
    Path ckpt = Paths.get(a.length > 4 ? a[4] : "as2kivi.done");
    Set<Integer> done = new HashSet<>();
    if (Files.exists(ckpt)) for (String l : Files.readAllLines(ckpt)) if (!l.isBlank()) done.add(Integer.parseInt(l.trim()));

    try (AerospikeClient as = new AerospikeClient(a[0], Integer.parseInt(a[1]));
         JedisPooled kv = new JedisPooled(a[3])) {     // rediss://... for TLS
      Statement stmt = new Statement();
      stmt.setNamespace(ns);                            // no set, no filter: every record
      long count = 0;
      for (int p = 0; p < 4096; p++) {
        if (done.contains(p)) continue;
        long now = System.currentTimeMillis() / 1000;
        try (Pipeline pipe = kv.pipelined();
             RecordSet rs = as.queryPartitions(null, stmt, PartitionFilter.id(p))) {
          while (rs.next()) {
            Key key = rs.getKey();
            Record rec = rs.getRecord();
            byte[] id = key.userKey == null
                ? ("#" + HexFormat.of().formatHex(key.digest)).getBytes()   // the key was not stored
                : key.userKey.getObject() instanceof byte[] b ? b : key.userKey.toString().getBytes(StandardCharsets.UTF_8);
            byte[] prefix = (ns + ":" + (key.setName == null ? "" : key.setName) + ":").getBytes(StandardCharsets.UTF_8);
            byte[] name = new byte[prefix.length + id.length];
            System.arraycopy(prefix, 0, name, 0, prefix.length);
            System.arraycopy(id, 0, name, prefix.length, id.length);
            Map<byte[], byte[]> fields = new HashMap<>();
            rec.bins.forEach((bin, v) -> {
              if (!(v instanceof Value.HLLValue)) fields.put(bin.getBytes(StandardCharsets.UTF_8), field(v));
            });
            if (fields.isEmpty()) continue;
            pipe.del(name);
            pipe.hset(name, fields);
            int ttl = rec.getTimeToLive();              // -1: never expires
            if (ttl > 0) pipe.expireAt(name, now + ttl);
            if (++count % 500 == 0) pipe.sync();        // bound what is in flight
          }
          pipe.sync();
        }
        Files.writeString(ckpt, p + System.lineSeparator(), StandardOpenOption.CREATE, StandardOpenOption.APPEND);
      }
      System.out.println("copied " + count + " records");
    }
  }
}

java -cp "$CLASSPATH" AsToKivi 10.0.0.5 3000 app "redis://:$KIVIDB_PASSWORD@kividb.internal:6380"

Floats are written in Java's format (1.0E7 where Python writes 10000000.0); both read back as the same number.