Forem: Aerospike

Introducing Aerospike Graph Database 3.0: Faster, simpler, and built for the terabyte scale era

Aerospike — Tue, 29 Jul 2025 20:49:32 +0000

Discover Aerospike Graph 3.0, built for billion-scale speed, ease, and cost-efficiency in fraud, identity, and personalization graph workloads.

Author: Ishaan Biswas, Director of Product Management - Graph

Today, we’re announcing the release of Aerospike Graph Database 3.0, bringing major improvements across the three dimensions that matter most: developer ease with multi-property and native datetime support, 10x faster ingest performance, and up to 50% better cost efficiency through reduced storage footprint. Built for teams managing the most demanding graph workloads in identity resolution, fraud prevention, and real-time personalization, Aerospike Graph Database 3.0 is our biggest step forward in making high-scale, graph data fast, accessible, and affordable.

Easier to develop

Aerospike Graph Database 3.0 introduces two key capabilities that make modeling and querying graph data more natural and powerful:

Multi-property support

Vertices can now store multiple values under the same property key. This simplifies how you can model complex signal relationships, such as multiple timestamps, scores, or segments, without complex workarounds or nested structures.

Example: Store multiple tags on a user vertex
g.addV('User') .property('T.id', 'user-123') .property(VertexProperty.Cardinality.list, 'tag', 'frequent-buyer') .property(VertexProperty.Cardinality.list, 'tag', 'mobile-user') .property(VertexProperty.Cardinality.list, 'tag', 'high-LTV')

Query all tags for the user
g.V().has('id', 'user-123').values('tag') // Output: ['frequent-buyer', 'mobile-user', 'high-LTV']

Or query Users with a tag
.V().has(“tag”, “high-LTV”) // Output: [user-123]

This is especially useful in AdTech, identity graphs, fraud, and personalization graphs, where entities accumulate multiple attributes over time.

Native Datetime support

Timestamps are now supported as a first-class property type. You can store, filter, and compare temporal data directly with no need to convert or encode datetime fields manually.

Example: Add a timestamp to a session vertex

g.addV('Session') .property('id', 'sessionid-123456') .property('startedAt', datetime('2025-07-08T14:22:00Z'))
Query sessions that started in the last 24 hours

g.V().hasLabel('Session') .has('startedAt', gt(datetime('2025-07-07T14:22:00Z')))

These capabilities make the platform more expressive and developer-friendly, especially for time-based and signal-rich domains.

Faster to load

Aerospike Graph 3.0 ships with a re-architected bulk loader that dramatically reduces ingest time, especially at large scale.

In production-scale benchmark tests, we loaded a terabyte of data in under three hours, down from over 32 hours with version 2.6. That’s a >10x performance gain, with the same infrastructure.

This performance unlocks faster iteration and experimentation cycles, allowing teams to load large datasets quickly, validate changes, and refresh data frequently, without needing to scale up infrastructure or wait for a long time to see the results of their changes.

Cheaper to run

Aerospike Graph Database 3.0 also delivers substantial gains in storage efficiency and overall total cost of ownership.

Improvements to on-disk data layout reduce storage footprint by up to 50%, which means fewer nodes are required to store the same data. This directly reduces cluster size and infrastructure costs.

Operate a terabyte-scale graph for a fraction of the cost of any other graph database
Enjoy predictable cost scaling, with $/GB dropping as data volume increases

Combined with Aerospike’s Hybrid Memory Architecture and the ability to scale compute and storage independently via the Aerospike Graph Service, Aerospike Graph offers the most cost-efficient ways to run high-performance graph workloads, without the constraints of RAM-bound architectures.

Get started

Aerospike Graph Database 3.0 is:

Easier to build on - with multi-property vertex support and native datetime types
Faster to load - with >10x ingest speed for billion-scale datasets
Cheaper to run - with smaller clusters, lower storage footprint, and reduced total cost of ownership at scale

If you’re solving large-scale graph problems that power real-time decisions, Aerospike Graph Database 3.0 is built for you.

→ [Read the release notes]

How Criteo scaled 290M QPS and cut its server footprint by 78%

Aerospike — Thu, 24 Jul 2025 21:36:16 +0000

Author: Steve Tuohy - Director of Product Marketing

Discover how Criteo scaled to 290M queries-per-second, swapped Couchbase + Memcached for Aerospike, cut servers by 78%, and kept sub-ms latency.

Criteo serves more than 700 million users daily with personalized ads, processing billions of events in milliseconds. At its peak, Criteo’s infrastructure handles 290 million key-value queries per second (QPS).

Until recently, this real-time engine ran on a complex stack of Couchbase and Memcached, propped up by custom C clients and costly operational overhead. Performance during rebalancing was fragile, cache warming required manual traffic shaping, and RAM costs were high.

But when Criteo rebuilt its stack using Aerospike’s patented Hybrid Memory Architecture (HMA), the company consolidated two systems into one, simplified operations, and cut its server footprint by 78%, all while maintaining sub-millisecond latency at global scale.

Maintaining real-time access at AdTech scale

Criteo isn’t just serving banner ads. It runs real-time auctions on the open web, responding to 20 million bid requests per second. Each request requires dozens of micro-decisions: audience scoring, campaign pacing, frequency caps, and more.

"Our key-value storage system peaks at about 290 million queries per second, which is fairly large,” said Maxime Brugidou, vice president of engineering, Criteo.

All of that happens in under 100 ms. The stack is written in C, runs on premises across 40,000 servers and seven data centers, and operates through Kubernetes and Apache Mesos.

The demands on storage? Sub-millisecond latency, no downtime during rebalancing or upgrades, and a globally distributed footprint.

The legacy stack: Couchbase + Memcached + custom C logic

Before adopting Aerospike, Criteo’s real-time infrastructure relied on Couchbase for the data, with Memcached as a caching layer. A custom-built C client orchestrated dual writes and replication between tiers, and maintained consistency between the two systems. “It was very difficult to make it perform well during rebalancing or maintenance,” Brugidou said. “It was quite unstable.”

Operationally, this design required careful tuning and careful rerouting of traffic. Any node failure or rebalance event degraded performance and had to be dealt with.

The turning point: Aerospike’s Hybrid Memory Architecture (HMA)

Aerospike’s HMA changed the game. It decouples index storage from data storage:

Indexes stay in RAM for fast access.
Data lives on SSDs, such as NVMe drives, in Criteo's case.

This model gave Criteo Memcached-level low latency with the durability of disk and let the company collapse its architecture from two databases into one. "There was this nice design with the index being in memory and the data on disk,” Brugidou said. “It allowed for really good tradeoffs."

By keeping indexes in memory and serving records from fast SSDs, Aerospike hit sub-millisecond reads without needing RAM to hold entire datasets. For Criteo, this not only simplified its system design but also saved money.

"Aerospike combined with NVMe disks… we had basically Memcached performance except that we were using persistent storage,” Brugidou said. “That was quite impressive…Aerospike was able to keep reading steadily at high throughput and very, very low latency despite all the mess we were putting on the servers."

Kubernetes-native deployment: From custom scripts to automated ops

Criteo replaced both Couchbase and Memcached with Aerospike. The company removed its custom client drivers, adopted Aerospike’s native C client, and rolled out a Kubernetes-native deployment on-prem with the Aerospike Kubernetes Operator.

Operational wins included:

Automatic node recovery and rebalancing via Kubernetes
Eliminating the custom C client, reducing complexity

Multi-bin optimization: Collapsing data models for performance

Aerospike’s multi-bin optimization simplified things even more. Criteo merged multiple datasets into one namespace, which helped reduce index memory usage and improve access efficiency. "You can write to some bins very easily and read all the bins at once. You only pay for indexing once in memory. That was quite a significant win," Brugidou said.

Aerospike also made rebalancing and node failure recovery seamless to the team. "We do that transparently,” Brugidou said. Aerospike is able to rebalance so easily… This is fully automated and very easy to do with the Kubernetes Operator.” Multi-bin optimization also meant:

Reduced index memory usage by avoiding duplication
Faster lookups, because all bins could be read in one fetch

This move simplified things even more, aligning with Criteo's infrastructure-as-code strategy and reducing human intervention during failure recovery.

Infrastructure impact: 78% server reduction and lower carbon footprint

By replacing Couchbase and Memcached with Aerospike, Criteo reduced server count from more than 3,000 to just 720. This meant:

Lowered total RAM and disk footprint
Reduced power consumption and cooling

Began running on 100% renewable energy in 2022, accomplished with the purchase of certificates and relocation of data centers to more sustainable locations.

“We’re doing many more queries for less electricity in the end. Way less,” said Brugidou. This migration supports Criteo’s science-based climate target to reduce electricity consumption by 42% by 2030.

Lessons learned: Performance tuning, not over-engineering

Despite the performance of the new system, Criteo didn’t jump straight to exotic optimizations. Instead, it prioritized simplicity:

Avoided pre-loading datasets in RAM
Let Aerospike handle replication and rebalancing
Tuned bin and namespace configs incrementally based on workload metrics

Brugidou emphasized that consistency and throughput remained stable even under stress tests, without resorting to low-level tuning.

Takeaway: Scaling real-time systems without scaling costs

By replacing a brittle, multi-tiered stack with Aerospike’s real-time engine, Criteo gained both technical efficiency and reduced costs. It consolidated two systems into one, improved performance, reduced latency, and slashed infrastructure overhead, without sacrificing resiliency or scale.

For teams building real-time systems with massive QPS and tight SLAs, Criteo’s experience shows what becomes possible with the right storage architecture.

With Aerospike, Criteo:

Consolidated two systems into one
Maintained sub-millisecond latency
Reduced infrastructure costs and energy use
Gained reliability through automation

Want to learn more? Check out the webinar replay or talk to our team about what real-time data infrastructure can do for you.

Aerospike Rust client preview: Introducing a native integration for high-performance systems

Aerospike — Thu, 17 Jul 2025 17:18:30 +0000

Brian Porter - Principal Product Manager, Database

Explore Aerospike's new official Rust client, built for memory-safe, low-latency systems. See features, roadmap, and how to start integrating today.
_

Developers building high-throughput, low-latency applications in Rust now have access to a high-performance client developed and maintained by Aerospike with the preview release of its official Rust client. This milestone represents more than just a new client library; it marks a shift from a community-supported project to one that business leaders can feel safe adopting.

Driven by increasing demand from our developer community and the rapid adoption of Rust for performance-critical, memory-safe systems, this release enables Rust developers to integrate Aerospike into their applications, all while preserving the consistency, speed, and reliability that define our platform.

Why Rust: Strategic alignment with developer needs

Rust is increasingly used across industries such as finance, telecommunications, cloud infrastructure, and embedded systems. It has become the preferred choice for performance-critical systems development. Organizations such as Amazon, Cloudflare, and Discord have adopted Rust to build reliable infrastructure and eliminate classes of memory-related bugs that commonly affect C/C++-based software.

Its popularity is driven by a combination of performance and reliability characteristics that are a natural fit for Aerospike applications:

No garbage collection: Rust’s lack of a garbage collector helps eliminate latency outliers, a key requirement for real-time systems.
Compile-time memory safety: Rust’s memory safety model reduces crash potential by catching common security vulnerabilities at compile time.
Safe concurrency support: Rust prevents data races by enforcing thread safety at compile time, helping high-concurrency applications scale better.
Modern tooling and developer experience: Over 100,000 Rust crates exist today, providing straightforward integration with Kafka, gRPC, observability tools, and more, forming a mature ecosystem.

Rust client capabilities

Rust Preview highlights include:

Execution models:

Async-first: Built for non-blocking IO, powered by Tokio by default, with optional support for async-std
Sync support: Blocking APIs available using a sync sub-crate for flexibility in legacy or mixed environments

Advanced data operations:

Batch protocol: Full support for read, write, delete, and UDF operations through the new BatchOperation API
New query/scan support: Implements updated query protocols for improved consistency and performance, allowing pagination and limiting the number of returned records

Policy and expression enhancements:

Replica policies: Includes support for Replica, including PreferRack policy
Policy additions: New fields such as allow_inline_ssd, respond_all_keys in BatchPolicy, read_touch_ttl, and QueryDuration in QueryPolicy
Rate limiting: Supports records_per_second for scan/query throttling

Data model improvements:

Type support: Adds support for boolean particle type
New data constructs: Returns types such as Exists, OrderedMap, and UnorderedMap, now supported for CDT reads
Value conversions: Implements TryFromAerospike::Value for type interoperability
Infinity and wildcard: Supports Infinity, Wildcard, and corresponding expression builders expressions::infinity() and expressions::wildcard()
Size expressions: Adds expressions::record_size() and expressions::memory_size() for granular control

Take a look at the changelog for more details.

The client has been built from the ground up to integrate easily with today’s Rust frameworks and async runtimes. Support for the Tokio async runtime, standard Rust crates, and idiomatic usage patterns are all part of the design.

Release timeline

The client is currently available as a preview release, with support for foundational operations. The general availability (GA) release is planned for later in 2025 and will include:

Full feature parity with other official Aerospike clients (e.g., Go, C#, Java)
Community feedback-driven improvements

The preview release is available on GitHub, and we encourage developers to experiment, provide feedback, and report issues.

Designed for the community
Aerospike developed this client because we listened to our developer community. Requests for async support, tighter performance tuning, and alignment with Rust’s ecosystem have been recurring themes in our forums and GitHub discussions. The new client is a direct response to that input.

By releasing this client under an open license and hosting it on GitHub, we aim to foster ongoing community engagement. Contributions, feature requests, and feedback will help shape the GA version and future enhancements.

Get started

Start working with the preview client today by going to the Installation section of the v2 branch of the Rust client repo.

Pi day exercise: Calculating pi with expressions!

Piyush Gupta — Thu, 14 Mar 2024 13:36:06 +0000

The Monte Carlo method is an effective technique for calculating probabilities.

In this fun blog, whose concept was "engineered" by our very own Joshua Pollack, we demonstrate how to use expressions in Aerospike to calculate the value of Pi using a Monte Carlo technique.

While the example is fun and rather contrived, it highlights an underlying technique of using expressions in Aerospike to compute incoming data values for thresholds and apply conditional executions to real-time situations such as fraud detection.

What is Pi?

Pi (often represented as the Greek letter π) is “one of the most well-known mathematical constants [and] is the ratio of a circle’s circumference to its diameter. For any circle, the distance around the edge is a little more than three times the distance across. Typing π into a calculator and pressing ENTER will yield the result 3.141592654, not because this value is exact, but because a calculator’s display is often limited to 10 digits.”

Calculating the value of Pi

The Monte Carlo method to compute the value of Pi uses the probability of a random point whose x and y coordinates are between -1.0 and 1.0 falling within the enclosing square of side 2.0 versus within the circle at center (0.0, 0.0) with radius r = 1.0.

The probability of being in the circle versus the encompassing square is π*r²/(2r)². This article describes this simple technique quite nicely.

The technique reduces to:

Estimated value of Pi = 4 * number of points inside the circle / total number of points.

We can apply the ratio to just the top quadrant of the circle and rectangle.

We can use the Java random number generator to produce x and y coordinates of a point between 0.0 and 1.0.

We will do this computation in a single Aerospike record that we initialize as follows:

AerospikeClient client = new AerospikeClient("127.0.0.1", 3000);

//Initialize our single aggregation record, center of circle 

Key key1 = new Key("test", "points", 1); 
Bin xcoord = new Bin("x", 0.0);
Bin ycoord = new Bin("y", 0.0);
Bin inCount = new Bin("inCount", 1.0);
Bin totalCount = new Bin ("count", 1.0);
WritePolicy wPolicy = new WritePolicy();
client.put(wPolicy, key1, xcoord, ycoord, inCount, totalCount );
//Check
System.out.println(client.get(null, key1));

Output:

(gen:1),(exp:448491015),(bins:(x:0.0),(y:0.0),(inCount:1.0),(count:1.0))

As we insert the next random values of the (x,y) pair, we will use expressions to compute whether the value lies within the circle or outside and in the same write transaction, update the inCount and the total count.

We use another expression-based call to get the estimated value of Pi (expected to be 3.14xx).

//Insert a record using a class based construct
class Point {

  public void insertPoint (Key key1, double xval, double yval, boolean bCheck) {
    Bin xcoord = new Bin("x", xval);
    Bin ycoord = new Bin("y", yval);
    WritePolicy wPolicy1 = new WritePolicy();

    Expression inCircleExp = Exp.build(
       Exp.cond(
        Exp.le(
          Exp.add(
            Exp.mul(Exp.floatBin("x"), Exp.floatBin("x")), 
            Exp.mul(Exp.floatBin("y"), Exp.floatBin("y"))
          ),
          Exp.val(1.0)
        ),
       Exp.add(Exp.floatBin("inCount"), Exp.val(1.0)),
       Exp.floatBin("inCount")
       ));

    Expression totalCountExp = Exp.build(Exp.add(Exp.floatBin("count"), Exp.val(1.0)));

    Record record = client.operate( wPolicy1, key1,   
          Operation.put(xcoord),
          Operation.put(ycoord),                   
          ExpOperation.write("inCount", inCircleExp, ExpWriteFlags.DEFAULT),
          ExpOperation.write("count", totalCountExp, ExpWriteFlags.DEFAULT)
                                  );                 
    if(bCheck){
      System.out.println(client.get(null, key1));
    }   
  }

  public void getPi(Key key1){
      WritePolicy wPolicy = new WritePolicy();
      Expression piExp = Exp.build(
          Exp.mul(Exp.val(4.0), 
                  Exp.div(Exp.floatBin("inCount"),Exp.floatBin("count"))
                 ));

      Record record = client.operate(wPolicy, key1, ExpOperation.read("pi", piExp, ExpReadFlags.DEFAULT));
      System.out.println("Pi : " + record.getValue("pi"));
  }

}

Let's try this code construct:

//Check
Point p = new Point();
p.insertPoint(key1, 0.71, 0.71, true);  //Out of circle, in rectangle

Output:

(gen:2),(exp:448490510),(bins:(x:0.71),(y:0.71),(inCount:1.0),(count:2.0))

Let's scale it up and insert 10,000 random points. This takes an imperceptible amount of time on Aerospike.

//Insert random points with x,y between 0.0 and 1.0

Random rnd1 = new Random(0);  

for(int i=1;i<10000;i++){
  double xval = rnd1.nextDouble();
  double yval = rnd1.nextDouble();
  p.insertPoint(key1, xval, yval, false);  
}

And now for our estimate of Pi:

p.getPi(key1);

Output:

Pi : 3.1424

Happy Pi Day!

In the above example, getPi() is shown here as a separate call. However, it can be made part of the return value in the write api (insertPoint()).

Instead of our pair of x and y, if you had 100s of such parameters that you would want to compute a net score to make a go/nogo decision, the above code constructs can be easily extended. In this example, we had no need to retain the 10,000 random values. However, if the data needs to be retained, consider using one of our collection data types, such as a Map or a List.

Get the most out of your data with Aerospike Expressions

To learn more, watch our Optimizing query performance with Aerospike Expressions webinar with Chief Developer Advocate Tim Faulkes.

How to retrieve the key in a key-value store

Piyush Gupta — Wed, 13 Mar 2024 17:54:01 +0000

Learn how to retrieve user-specified keys in Aerospike using single record reads.

In this blog, you’ll learn how to retrieve the user-specified key in Aerospike, a key-value store.

Aerospike is designed from the ground up to support extremely low latency create, read, update, and delete (CRUD) operations. In Aerospike, applications store data as individual records that are uniformly distributed across all the nodes of the database cluster. A user-specified key is associated with each record along with the application data. Below, we will walk you through the steps necessary to retrieve your user-specified key in a single record read operation.

Keys in Aerospike

As mentioned above, Aerospike stores user data in individual records, which are accessible by a user-specified "key." Keys can only be a string, integer, or a byte array.

[USER SPECIFIED KEY] → [USER DATA in one or more "bins"]

Applications access the data via the Aerospike client library, which computes a 20-byte “digest” by hashing the user-supplied key information of set name, user key value, and user key type.

The client library sends the 20-byte digest as the key of record to the server. So, the record in the server should read:

[20 Byte Digest] → [USER DATA in one or more "bins"]

Storing keys in records

What if two keys hash to the same digest value, resulting in a hash collision?

Aerospike uses the RIPEMD-160 hashing algorithm, which has a statistically near-zero probability of hash collision. By default, Aerospike does not store the user-supplied key with the saved record. However, applications have the option to store the user-specified key along with the record via the write policy.

[20 Byte Digest ] → [USER DATA in one or more "bins"] + [User Key (string / int / blob)]

The approach above ensures that the server verifies the hash against the stored key during CRUD operations. Any subsequent updates where a hash collision is detected, the application will receive a KEY_MISMATCH error.

Key mismatch explained

Consider the case in a namespace test, where a user-specified key of abc in set testSet1 generates a particular digest, digest1. When this record is first created, Aerospike uses digest1 as the internal key of that record but also stores the user-specified key of abc with the record.

Hypothetically, in the same namespace test, consider the user-specified key of xyz in set testSet2, which also happens to hash to the same value, digest1. (This has near zero probability, practically speaking.)

If the application tries to create or update a record with digest1 and with user-specified key xyz, Aerospike will access the record stored against digest1 and compare the stored key abc against the incoming transaction's user-specified key of xyz and discover that they are different. A KEY_MISMATCH exception will be thrown in this case.

User key stored with the record

The curious may want to know: Is it possible to read this user-specified stored key back in the application?

For data modeling purposes and for queries (primary or secondary index queries that retrieve qualifying records from the entire namespace), the user stored key can be retrieved back to the client. This is useful since the user may want to then update individual records selectively, as part of their use case.

However, for single key reads the user key is not returned to the client, as you must have had the key to be able to read the record in the first place. This type of read can be done by passing either the user key and the set name or the digest.

In certain operational situations, users have asked if it is possible to read back the user stored key on an individual record read. They stored their key with the record and would now like to retrieve it. These are situations where users have the 20-byte digest of a record, but they do not know the user key. The digest was recorded in the server logs. An example of this would be when turning on rw-client logging to identify the digests of "hot-keys," records that are being updated very frequently.

The read APIs don't provide this functionality upfront via the policies since this is not a typical data modeling need. However, Aerospike does provide expressions that can access the user stored key.

What are Aerospike Expressions?

An expression is a strongly typed domain-specific language designed for manipulating and comparing bins and record metadata. There is no “programming language” per se; an expression is built with code. It runs on the nodes in the cluster, not on the client.

Aerospike supports three types of expressions:

Filter expressions (introduced in version 5.2.0)
Cross Datacenter Replication filter expressions (introduced in version 5.3.0)
Operation expressions (introduced in version 5.6.0)

Filter and operation expressions can be specified on a per-call basis.

Key retrieval code example

Below is an example code snippet demonstrating how to retrieve a key using the Java client.

Creating the record initially:

AerospikeClient client = new AerospikeClient("127.0.0.1", 3000);
Key key = new Key("test", "testset", "key1");
WritePolicy wPolicy = new WritePolicy();

wPolicy.sendKey = true;  //Store our user key on the server

Bin b0 = new Bin("b00", Value.get(28));
client.put(wPolicy, key, b0);

Now, assume we just have the 20-byte digest and would like to retrieve our key:

We can use the following key constructor:

// Initialize key from namespace, digest, optional set name, and optional userKey.
Key(String namespace, byte[] digest, String setName, Value userKey)

Let's find the digest of the record we created above, for purposes of this demo.

byte[] recDigest = new byte[20];
recDigest =  Key.computeDigest("testset", Value.get("key1")) ;

Next, create a new key object using the digest-based constructor.

Key keyByDigest = new Key("test", recDigest, null, Value.NULL);

About `Value.NULL`

The constructor allows specifying the user-specified key in the last argument as a “Value" type object. For example, Value.get("myKey"). This is provided for the rare case where the application wants to do an update with the key generated using the digest, and the set name is not known but still wants to update with sendKey=true. In our case, we don't know the key and are trying to retrieve it in an operate() call, which is like a write, although we will only perform a read via expressions. Hence, we pass Value.NULL for the user-specified key and use sendKey=false.

We should be able to read the same record using either "key" or "keyDigest." Let's validate.

System.out.println("Record: "+ client.get(null, key)); 
System.out.println("Record via Digest: "+ client.get(null, keyByDigest));

Output:

Record: (gen:12),(exp:447740032),(bins:(b00:28))
Record via Digest: (gen:12),(exp:447740032),(bins:(b00:28))

Case 1: key type known

Now, using the digest-based key, let's retrieve the stored key from the server. Since we are specifying the digest and Value.NULL for the userKey in our recDigest constructor, we must set thewPolicy.sendKey = false; to avoid a KEY_MISMATCH error.

wPolicy.keySend = false;
Expression recKeyExp = Exp.build(Exp.key(Exp.Type.STRING));
Record record = client.operate( wPolicy, keyByDigest,   
          ExpOperation.read("reckey", recKeyExp, ExpReadFlags.DEFAULT) 
         );
System.out.println("Record Key: " + record.getValue("reckey"));

Recall we mentioned that user-specified keys can be of type integer, string, or byte-array. Here, although we don't know our key, we know it is a string type. This is specified using Exp.Type.STRING when building the expression for the key.

Output:

Record Key: key1

Case 2: key type unknown

What if we don't know the key type? We can check for all the three allowable types. Although, in this case, we must use ExpReadFlags.EVAL_NO_FAIL instead of ExpReadFlags.DEFAULT so that the failed expressions with type mismatch don't generate an Exception.

Example code is shown below.

wPolicy.sendKey = false;
Expression recKeyExpStr = Exp.build(Exp.key(Exp.Type.STRING));
Expression recKeyExpInt = Exp.build(Exp.key(Exp.Type.INT));
Expression recKeyExpBlob = Exp.build(Exp.key(Exp.Type.BLOB));

Record record = client.operate( wPolicy, keyByDigest,   
          ExpOperation.read("reckeyStr", recKeyExpStr, ExpReadFlags.EVAL_NO_FAIL),
          ExpOperation.read("reckeyInt", recKeyExpInt, ExpReadFlags.EVAL_NO_FAIL),
          ExpOperation.read("reckeyBlob", recKeyExpBlob, ExpReadFlags.EVAL_NO_FAIL)  
         );
System.out.println("Record Key String: " + record.getValue("reckeyStr"));
System.out.println("Record Key Integer: " + record.getValue("reckeyInt"));
System.out.println("Record Key BLOB: " + record.getValue("reckeyBlob"));

Output:

Record Key String: key1
Record Key Integer: null
Record Key BLOB: null

Unlock the power of expressions in Aerospike for enhanced data operations

Expressions are a very powerful tool in Aerospike. They can be used as filters for selecting records to return in primary or secondary index queries, as well as compute on a record’s existing data or metadata. It can then return the result of the computation as a read or write it to a new bin. The possibilities are only limited by your creativity!

To learn more, watch our Optimizing query performance with Aerospike Expressions webinar with Chief Developer Advocate Tim Faulkes.

Using Aerospike policies (correctly!)

Tim F — Thu, 29 Feb 2024 23:28:25 +0000

Developers familiar with Aerospike will know that the vast majority of the API calls take a “policy” as a parameter. This policy is optional but can have a significant impact on the application. There are a number of parameters in policies that can be confusing to developers, and even creating the policies is frequently done incorrectly. This blog will attempt to clear up some of this confusion.

What are policies?

Each API call in Aerospike typically takes a policy as a parameter. For example, consider a simple get in Java:

client.put(null, new Key("test", "testSet", 1), 
           new Bin("name", "Tim"), new Bin("age", 312));

The first parameter here is the policy; in this case, it is an instance of the WritePolicy class. If null is specified, then the default policy is used. Policies typically contain fields that fall into one of three categories:

Networking control: How should the API call behave if something goes wrong? For example, if the server cannot be reached, should the call be retried on a different server?
Application behavior: Parameters that control how the API call should behave in various application-driven scenarios. For example, Aerospike does an “upsert” operation by default on a put, selecting to either update or insert the record depending on whether it already exists or not. However, the application might require the record to be inserted only, failing if the record already exists in the database.
Filtering: Aerospike supports filters on operations, allowing almost all operations to become conditional. For example, an operation can be specified to retrieve a record if the amount bin in that record is greater than a thousand.

These three categories are all in the same policy object, with the networking control defined on the Policy class, which is a superclass of other policies, such as WritePolicy. Note that the Policy class is used for read operations without a subclass; there is no ReadPolicy class.

Networking control policies

The policy superclass defines a number of parameters related to networking. These are critical to optimizing application behavior but are often poorly understood. Let’s take a look at these and the interaction between them.

These networking fields are:

connectTimeout
maxRetries
replica
sleepBetweenRetries
socketTimeout
timeoutDelay
totalTimeout

A diagram is helpful in understanding the interaction between these fields:

Figure 1: API call sequence

The diagram shows a timeline of what happens when an API call is submitted to the Aerospike Client Library (ACL) and can be read from left to right.

The first thing that happens is the ACL needs a connection to the server which holds the data. If this is a simple get or put, this will be a connection to a single server, but if it’s a complex operation such as a batch or query, multiple servers might be involved. The ACL has connection pools to each server node, so acquiring this connection is typically a matter of borrowing one from the pool, which is almost instantaneous. However, if the pool has no available connections, a new connection must be established, which can take some time. This is especially true if the connection is over TLS as the handshake between the client and server on a TLS connection can be slow.

This connection establishment process has its own timeout field, connectTimeout. This field is in milliseconds and includes both the connection establishment as well as user authentication (if any). By default, this is set to zero, which is usually a reasonable default – this uses the lesser of the socketTimeout and totalTimeout, or 2,000 milliseconds if both are zero. If the application can tolerate extra time when a connection is established, especially when using TLS, then this can be set to a non-zero value. For example, if you want a large connectTimeout and a small socketTimeout to allow for protracted SSL handshakes but fast application responses, setting this value to something like 5,000 milliseconds works well.

Once a connection has been retrieved, the API call is sent to the server. There are two timeouts compared at this point, the socketTimeout and the totalTimeout. Both are in milliseconds, and the socketTimeout reflects how long it takes to wait for data from this API call before retrying or failing. The totalTimeout specifies the maximum time the call can take, including retries. If the time remaining in the totalTimeout is less than the socketTimeout, then the totalTimeout is used as the socketTimeout for the call.

In Figure 1, the socketTimeout is less than the totalTimeout, so the ACL waits until one of the following occurs: socketTimeout milliseconds have elapsed, the connection returns a timeout, or the call succeeds. If the call was unsuccessful in this period, the ACL checks the maxRetries setting to see if any retries remain. For reads, this value defaults to two, giving it a total of three tries – the initial attempt plus the two retries. For writes, this value defaults to zero, so no retries are attempted.

Figure 1 depicts a read with the default two retries, so once the original socketTimeout expires, the API call does not return to the application. Instead, the ACL waits for sleepBetweenRetries milliseconds and then submits the call again.

This second call may or may not go to the same server. This is controlled by the replica policy setting, which defaults to SEQUENCE. Sequence means that the first attempt is made to the node containing the master record, but on failure, the retry is made to a replica. This can be very useful in a read-sensitive use case, for example. If the socketTimeout is set to a short interval (say 20 milliseconds) and there are retries with a replica policy of SEQUENCE if the master node is busy and cannot respond in time or has fallen over. Still, if the cluster has not detected this yet, the retry will go to the replica and may return data within the application’s SLA.

If this second call again times out after socketTimeout milliseconds, a second delay of sleepBetweenRetry milliseconds is performed then the second retry (3rd total attempt) is performed.

Let’s take a look at where this leaves us in Figure 1:

The ACL is about to start the second retry, but in this case, the remaining time in the totalTimeout is less than the socketTimeout. Hence the API call will timeout before the socketTimeout elapses when the remaining time in the totalTimeout has elapsed.

Note that once the totalTimeout has elapsed, the client API call fails, even if further retries are available.

Timeout delay

The only timeout setting not covered in the above is the timeoutDelay setting. This is not strictly related to the API behavior but rather the system's efficiency. As the above shows, when the ACL submits the call to the server node, and no response is received within socketTimeout milliseconds, either the call is abandoned, or a retry is performed. But what happens to the connection on which that original request was placed?

By default, that connection is closed, destroying the connection and not placing it back in the connection pool. This is because it is possible that the server simply did not respond in time, and it might respond some time later. If the connection was reused instead of being closed, this server response might confuse the communication of the reusing call.

However, as discussed above, connection establishment can be expensive. Closing connections typically means that this connection must be re-established later, which can impact the application.

The timeoutDelay parameter specifies a time interval in milliseconds for those connections to be kept alive. If the server responds within that timeout, the socket is drained and then returned to the connection pool. If the timeout is exceeded, then the connection is closed.

Note that there is a cost of using the timeoutDelay as the ACL must keep track of which connections have timed out. It is useful in situations where timeouts may occur frequently (for example, aggressive read settings) and connection establishment is expensive.

Default policies

Now that the timeout settings are understood, we can discuss default policies. The ACL defines default values of the various policy attributes, which users can customize to suit their application instead of repeatedly modifying the defaults in the application code. Default policies are set up when the ACL is created and are used when null is passed to the policy parameter in the API call. Most often, the default policies are customized with the network settings discussed above.

Here is an example which creates a default writePolicy with a socketTimeout of 2,000 milliseconds:

WritePolicy writePolicyDefault = new WritePolicy();
writePolicyDefault.socketTimeout = 2000;

ClientPolicy clientPolicy = new ClientPolicy();
clientPolicy.writePolicyDefault = writePolicyDefault;
IAerospikeClient client = new AerospikeClient(clientPolicy, "127.0.0.1", 3000);

Note that the writePolicyDefault is set on the ClientPolicy instance before the ACL is created. When passed to the ACL constructor, a copy of the default policies is made so no changes are possible to the default policies.

After executing this code, passing null to a write API call will result in the ACL retrieving and using the defaultWritePolicy. Hence, the following two calls are identical:

client.put(null, key, new Bin("name", "joe"));
client.put(client.getWritePolicyDefault(), key, new Bin("name", "joe"));

Creating policies

Timeout settings are typically set on default policies and reused multiple times across the application. However, application-level settings are done as needed by the application on a per-call basis. Consider a check-and-set scenario where a record has been read from the database, changes made, and then those changes pushed back to the database, but only if the record has not been changed since it was read:

Record record = client.get(null, key);
...
WritePolicy writePolicy = client.getWritePolicyDefault();
writePolicy.generation = record.generation;
writePolicy.generationPolicy = GenerationPolicy.EXPECT_GEN_EQUAL;
client.put(writePolicy, key, new Bin("name", "joe"));

This code attempts to solve the requirements, getting the defaultWritePolicy from the ACL and then setting the generation and generationPolicy correctly. However, this code is very dangerous and will create issues in the application. There is just one defaultWritePolicy on the ACL, and the getWritePolicyDefault()call will return a reference to that object, not a copy of that object. Hence, setting the generation values on that policy will affect all calls that use that default policy on all threads, almost certainly causing many calls to fail with an exception.

Another common pattern to solve this problem is:

WritePolicy writePolicy = new WritePolicy();
writePolicy.generation = record.generation;
writePolicy.generationPolicy = GenerationPolicy.EXPECT_GEN_EQUAL;
client.put(writePolicy, key, new Bin("name", "joe"));

While this code is an improvement on the previous snippet, it too, has a problem. In this case, a unique WritePolicy instance has been created, preventing issues with this call affecting other calls. However, using new WritePolicy() does not pick up the settings of the default write policy, so settings changed on this default policy, such as the socketTimeout will not apply to this new policy.

The correct way to create a new policy to override fields is as follows:

WritePolicy writePolicy = new WritePolicy(client.getWritePolicyDefault());
writePolicy.generation = record.generation;
writePolicy.generationPolicy = GenerationPolicy.EXPECT_GEN_EQUAL;
client.put(writePolicy, key, new Bin("name", "joe"));

In this case, a new WritePolicy is created, but it gets a copy of all the settings off the default write policy. This object is local to this API call, so there is no contention with other calls.

Summary

Aerospike’s Policy classes are both powerful and flexible, allowing fine-grained, per-call control over both network and application-level settings. However, it is important to understand what the settings do to correctly control the API call. Network settings are often overlooked during the application development phase and are frequently poorly understood. A thorough understanding of them is critical to developing applications that perform well not just in the happy-day scenario but also in the face of network issues or server failures.

Developers who are new to Aerospike should understand this simple pattern to create new policies. It will ensure the correct execution of the API without affecting other concurrent calls and unintended surprises.

Unlock Aerospike Cloud: Exploring the power of the Aerospike LINQPad driver

Richard Andersen — Wed, 28 Feb 2024 02:05:46 +0000

This blog is part of a continuing series showing how to use the Aerospike LINQPad driver. For other blogs in this series, please see:

The Aerospike LINQPad driver is now able to connect to the Aerospike Cloud, providing all the power of LINQPad combined with the Aerospike LINQPad driver.

This includes a graphical interface, LINQ support, and enhanced driver features to support the Aerospike Database. Find more information about LINQPad here.

Let’s look at the new connection dialog and some of the features of the LINQPad driver.

Connection dialog

The new connection dialog will now have two tabs. The first tab will be for native, “self-managed”, Aerospike clusters, and the second tab is used to connect to Aerospike Cloud.

Below shows the connection properties:

Aerospike Cloud connection dialog

A review of each property is below:

The host name is provided in the cloud dashboard. The hyperlink will take you to the Aerospike login screen or, if you are already logged in, to the dashboard.
The cloud client connection port.
If a VPC (AWS private link) is defined, this would be the host name displayed on the cloud dashboard. The host name field would be the VPC endpoint.
The API key is created via the dashboard.
If the API key and secret were exported from the dashboard to your local machine, this button will allow you to import that API Key CSV file. Properties “API Key” (item 4) and “API Secret” (item 7) are not required.
If enabled, the built-in LINQPad password manager is used. When enabled, a dropdown will be displayed to select the Password Name you defined in the manager. The Password Manager can be found on the File menu of LINQPad.
The associated API Key’s secret.
If checked, it will show the API secret in plain text.
The associated cloud namespace.
If provided, a list of set names separated by a comma or space. The set names will be used to populate the sets under the namespace in the LINQPad connection tree (see below image). Also, set and bin “detection” will be performed to obtain the bins and data types. Regardless of whether this field is provided, you can always obtain this information from the “Null Set.” See the “Using Null Set” section below.
Hyperlinks to additional topics.
The timeout values that will be used when obtaining the connection or performing an operation. Note that the “Sleep” field is ignored for cloud connections.

Below shows the relationship between the “set names” connection property and the LINQPad connection tree:

Association between the “Set Names” property and the sets populated in the connection tree

Driver overview

All the features of the Aerospike LINQPad driver are available, including the following:

Graphical interface with all LINQPad features
Drag-and-drop of set names
LINQ
Serialize and deserialize any C# object via the Object-Mapper (POCO)
Auto-Values, which provide dynamic data type conversion from Aerospike types to .Net types without testing or casting
JSON support
Explore the Aerospike API directly or use the driver’s enhanced API

Demonstrates some of the capabilities of LINQPad and the LINQPad driver

Using the Aerospike null set

If the “Set Names” connection property is not provided, the set and bin names will not be provided in the LINQPad connection tree (see image above).
You can still perform queries and API calls using the driver’s API features. Below is an example of a driver API call to obtain the records for Aerospike set “Artist”:

aerospike_cloud.NullSet.Where(ns => ns.Aerospike.SetName == "Artist")

Below is the output:

Result using the driver’s extended API for set “Artist”

You can always extend the “where” clause or interact with the result set.

Samples

You can find samples for Aerospike LINQPad Cloud under the samples folder or under the “Samples” tab in LINQPad.

Review

In this installment of the Aerospike LINQPad driver blog series, we learned how to use LINQPad to connect to Aerospike Cloud. This connection makes exploring your data in the cloud just as easy as other deployment options. Stay tuned for more articles in this series highlighting additional features of the driver.

How to use the Kubernetes Operator to establish connectivity between clusters

Naresh Maharaj — Fri, 09 Feb 2024 16:32:21 +0000

Discover how Aerospike's Cross Datacenter Replication feature (XDR) transfers data between clusters effortlessly.

This post outlines how to establish connectivity between two Aerospike clusters by leveraging Aerospike's Cross Datacenter Replication (XDR) feature to seamlessly transmit data from a source cluster to a destination cluster, providing high availability for critical data systems.

Ensuring network visibility for Aerospike service ports in the remote data center from a source cluster is crucial. However, this can pose challenges, especially in a Kubernetes environment. Fortunately, there is a solution: deploying a proxy server in front of the private Kubernetes destination cluster. To showcase this solution, we recommend installing the Kubernetes Operator to facilitate the creation and scheduling of the source and destination databases. While the demonstration is on setting up replication in one direction, it's worth noting that Aerospike supports active-active replication. This is accompanied by a conflict resolution mechanism to handle update clashes. This XDR proxy also supports this feature as well.

Let's begin the process.

Fig. 1 Aerospike Kubernetes Operator

The following Kubernetes nodes have been created using EKS.

First, display the following private and public IP addresses from listing the nodes with the kubectl command.

kubectl get nodes -o wide

NAME                             STATUS   ROLES    AGE     VERSION                INTERNAL-IP      EXTERNAL-IP      OS-IMAGE         KERNEL-VERSION                 CONTAINER-RUNTIME
ip-192-168-11-132.ec2.internal   Ready    <none>   2m53s   v1.22.15-eks-fb459a0   192.168.11.132   44.201.67.177    Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.17
ip-192-168-31-131.ec2.internal   Ready    <none>   2m52s   v1.22.15-eks-fb459a0   192.168.31.131   44.192.83.79     Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.17
ip-192-168-41-140.ec2.internal   Ready    <none>   2m51s   v1.22.15-eks-fb459a0   192.168.41.140   18.208.222.35    Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.17
ip-192-168-41-63.ec2.internal    Ready    <none>   2m51s   v1.22.15-eks-fb459a0   192.168.41.63    54.173.138.131   Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.17
ip-192-168-59-220.ec2.internal   Ready    <none>   2m52s   v1.22.15-eks-fb459a0   192.168.59.220   54.227.122.222   Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.17
ip-192-168-6-124.ec2.internal    Ready    <none>   2m51s   v1.22.15-eks-fb459a0   192.168.6.124    35.174.60.1      Amazon Linux 2   5.4.219-126.411.amzn2.x86_64   docker://20.10.1

Now, get a copy of the Aerospike git repo for the Kubernetes Operator.

git clone \ https://github.com/aerospike/aerospike-kubernetes-operator.git
cp features.conf \ aerospike-kubernetes-operator/config/samples/secrets/.

Setup

Run the following commands in the order specified. Wait for the CSV “Succeeded phase” to appear after running this line. Initially, it might take between 30 seconds and a minute to show up.

kubectl get csv -n operators -w

cd aerospike-kubernetes-operator/
kubectl apply -f config/samples/storage/eks_ssd_storage_class.yaml
curl -sL \
https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.22.0/install.sh | bash -s v0.22.0
kubectl create -f \
https://operatorhub.io/install/aerospike-kubernetes-operator.yaml
kubectl get csv -n operators -w
cd..
git clone \ https://github.com/nareshmaharaj-consultant/kubernetes-anything
cd kubernetes-anything
./createNamespace.sh aerospike
cd ../aerospike-kubernetes-operator/
kubectl -n aerospike create secret generic aerospike-secret \
--from-file=config/samples/secrets
kubectl -n aerospike create secret generic auth-secret \
--from-literal=password='admin123'

Destination Cluster

Use the following YAML configuration file for our destination cluster. Save the file and name it ssd1_xdr_dest_6.1_cluster_cr.yaml:

apiVersion: asdb.aerospike.com/v1beta1
kind: AerospikeCluster
metadata:
  name: aerocluster-dest-xdr
  namespace: aerospike

spec:
  size: 1
  image: aerospike/aerospike-server-enterprise:6.1.0.2

  storage:
    filesystemVolumePolicy:
      initMethod: deleteFiles
      cascadeDelete: true
    blockVolumePolicy:
      cascadeDelete: true
    volumes:
      - name: workdir
        aerospike:
          path: /opt/aerospike
        source:
          persistentVolume:
            storageClass: ssd
            volumeMode: Filesystem
            size: 1Gi
      - name: ns
        aerospike:
          path: /opt/aerospike/data/
        source:
          persistentVolume:
            storageClass: ssd
            volumeMode: Filesystem
            size: 1Gi
      - name: aerospike-config-secret
        source:
          secret:
            secretName: aerospike-secret
        aerospike:
          path: /etc/aerospike/secret

  podSpec:
    multiPodPerHost: true

  aerospikeAccessControl:
    roles:
      - name: writer
        privileges:
        - read-write
      - name: reader
        privileges:
        - read
    users:
      - name: admin
        secretName: auth-secret
        roles:
          - sys-admin
          - user-admin
          - read-write
      - name: xdr-writer
        secretName: xdr-user-auth-secret
        roles:
          - writer

  aerospikeConfig:
    service:
      feature-key-file: /etc/aerospike/secret/features.conf
    security: {}
    network:
      service:
        port: 3000
      fabric:
        port: 3001
      heartbeat:
        port: 3002
    namespaces:
      - name: test
        memory-size: 134217728
        replication-factor: 1
        storage-engine:
          type: device
          files:
            - /opt/aerospike/data/test.dat
          filesize: 1073741824
          data-in-memory: true

Next, create the following Kubernetes resources for our Aerospike destination cluster:

XDR destination database user login credentials, as a Kubernetes secret
Destination database cluster using our YAML file named ssd1_xdr_dest_6.1_cluster_cr.yaml

export secret_auth_name=xdr-user-auth-secret
export password_secret=admin123
kubectl -n aerospike create secret generic $secret_auth_name \
--from-literal=password=$password_secret
kubectl create -f ssd1_xdr_dest_6.1_cluster_cr.yaml
kubectl -n aerospike get po -w

You should see the database pods up and running successfully.

kubectl get po -n aerospike -w

NAME                       READY   STATUS     RESTARTS   AGE
aerocluster-dest-xdr-0-0   0/1     Init:0/1   0          13s
aerocluster-dest-xdr-0-0   0/1     Init:0/1   0          18s
aerocluster-dest-xdr-0-0   0/1     PodInitializing   0          19s
aerocluster-dest-xdr-0-0   1/1     Running           0          24s

XDR-Proxy

Next, set up the xdr-proxy. Take a brief look at Fig. 1 Aerospike Kubernetes Operator, and you’ll notice that we are working from the right to the left in that order.

Configuration

Create the following xdr-proxy configuration file. Replace the seed address with a fully qualified domain name (FQDN) for the destination database pod(s) you created earlier. Multiple seed addresses may be added (hint: recommended in production).

cd ..
mkdir -p xdr-cfg/etc/auth
cd xdr-cfg/etc/

cat <<EOF> aerospike-xdr-proxy.yml
# Change the configuration for your use case.
# Naresh Maharaj
# Refer to https://www.aerospike.com/docs/connectors/enterprise/xdr-proxy/configuration/index.html
# for details.

# The connector's listening ports, manage service, TLS, and network interface.
service:
  port: 8901
  # Aerospike Enterprise Server >= 5.0
  manage:
    address: 0.0.0.0
    port: 8902

# The destination aerospike cluster.
aerospike:
  seeds:
    - aerocluster-dest-xdr-0-0.aerospike.svc.cluster.local:
        port: 3000
  credentials:
    username: xdr-writer
    password-file: /etc/aerospike-xdr-proxy/auth/password_DC1.txt
    auth-mode: internal

# The logging config
logging:
  enable-console-logging: true
  file: /var/log/aerospike-xdr-proxy/aerospike-xdr-proxy.log
  levels:
    root: debug
    record-parser: debug
    server: debug
    com.aerospike.connect: debug
  # Ticker log interval in seconds
  ticker-interval: 3600
EOF

sudo tee auth/password_DC1.txt <<EOF
admin123
EOF
cd ..

kubectl -n aerospike create configmap xdr-proxy-cfg --from-file=etc/
kubectl -n aerospike create secret generic xdr-proxy-auth-secret \
--from-file=etc/auth

Deployment

Now that you have the xdr-proxy configuration file created, produce the Kubernetes deployment YAML file for the xdr-proxy. The following YAML file is used to deploy the xdr-proxy pods, ideally in the same data center or location where the destination databases will be hosted.

Tip: Remember that your xdr-proxy configuration differs from the Kubernetes xdr-proxy deployment file.

cat <<EOF> xdr-proxy-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: xdr-proxy
  namespace: aerospike
  labels:
    app: xdr-proxy
spec:
  replicas: 2
  selector:
    matchLabels:
      app: xdr-proxy
  template:
    metadata:
      labels:
        app: xdr-proxy
    spec:
      containers:
      - name: xdr-proxy
        image: aerospike/aerospike-xdr-proxy:2.1.0
        volumeMounts:
        - name: xdr-proxy-dir
          mountPath: "/etc/aerospike-xdr-proxy/"
          readOnly: true
        - name: xdr-auth-dir
          mountPath: "/etc/aerospike-xdr-proxy/auth"
          readOnly: true
        ports:
          - name: xdr-proxy-main
            containerPort: 8901
          - name: xdr-proxy-mng
            containerPort: 8902
      volumes:
      - name: xdr-proxy-dir
        configMap:
          name: xdr-proxy-cfg
          optional: false
      - name: xdr-auth-dir
        secret:
          secretName: xdr-proxy-auth-secret
          optional: false
---
apiVersion: v1
kind: Service
metadata:
  name: xdr-proxy
  namespace: aerospike
spec:
  selector:
    app: xdr-proxy
  ports:
  - name: main
    protocol: TCP
    port: 8901
    targetPort: xdr-proxy-main
  - name: manage
    protocol: TCP
    port: 8902
    targetPort: xdr-proxy-mng
EOF

kubectl create -f xdr-proxy-deployment.yaml
kubectl get po -n aerospike -w

The following command shows the current pods are running successfully. So far, so good!

kubectl get po -n aerospike -w
NAME                         READY   STATUS    RESTARTS   AGE
aerocluster-dest-xdr-0-0     1/1     Running   0          77m
xdr-proxy-7d9fccd6c8-g5mjt   1/1     Running   0          2m26s
xdr-proxy-7d9fccd6c8-mjxp4   1/1     Running   0          2m26s

Source cluster

Create the Aerospike source cluster using the following configuration. We will insert our sample messages here.

cd ../aerospike-kubernetes-operator/

cat <<EOF> ssd1_xdr_src_6.1_cluster_cr.yaml
apiVersion: asdb.aerospike.com/v1beta1
kind: AerospikeCluster
metadata:
  name: aerocluster-source-xdr
  namespace: aerospike

spec:
  size: 1
  image: aerospike/aerospike-server-enterprise:6.1.0.2

  storage:
    filesystemVolumePolicy:
      initMethod: deleteFiles
      cascadeDelete: true
    blockVolumePolicy:
      cascadeDelete: true
    volumes:
      - name: workdir
        aerospike:
          path: /opt/aerospike
        source:
          persistentVolume:
            storageClass: ssd
            volumeMode: Filesystem
            size: 1Gi
      - name: ns
        aerospike:
          path: /opt/aerospike/data/
        source:
          persistentVolume:
            storageClass: ssd
            volumeMode: Filesystem
            size: 1Gi
      - name: aerospike-config-secret
        source:
          secret:
            secretName: aerospike-secret
        aerospike:
          path: /etc/aerospike/secret

  podSpec:
    multiPodPerHost: true

  aerospikeAccessControl:
    roles:
      - name: writer
        privileges:
        - read-write
      - name: reader
        privileges:
        - read
    users:
      - name: admin
        secretName: auth-secret
        roles:
          - sys-admin
          - user-admin
          - read-write

  aerospikeConfig:
    service:
      feature-key-file: /etc/aerospike/secret/features.conf
    security: {}
    network:
      service:
        port: 3000
      fabric:
        port: 3001
      heartbeat:
        port: 3002
    xdr:
      dcs:
        - name: DC2
          connector: true
          node-address-ports:
            - xdr-proxy.aerospike.svc.cluster.local 8901
          namespaces:
            - name: test
    namespaces:
      - name: test
        memory-size: 134217728
        replication-factor: 1
        storage-engine:
          type: device
          files:
            - /opt/aerospike/data/test.dat
          filesize: 1073741824
          data-in-memory: true
EOF

kubectl create -f ssd1_xdr_src_6.1_cluster_cr.yaml
kubectl get po -n aerospike -w

From the source cluster, confirm the XDR component has connected to the xdr-proxy by filtering the Kubernetes log file as shown in the following kubectl command.

kubectl -n aerospike logs aerocluster-source-xdr-0-0 -c aerospike-server | grep xdr | grep conn
Dec 08 2022 13:49:21 GMT: INFO (xdr): (dc.c:581) DC DC2 connected Oct 10 2022 13:57:53 GMT: INFO (xdr): (dc.c:581) DC DC2 connected

Simple message test

Add some sample messages to the source database and confirm they are being received in the destination database cluster. Start by getting the source database service address and connect using Aerospike's command line tool - AQL.

kubectl get svc -n aerospike

NAME                         TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)             AGE
aerocluster-dest-xdr         ClusterIP   None             <none>        3000/TCP            3h38m
aerocluster-dest-xdr-0-0     NodePort    10.100.226.179   <none>        3000:30168/TCP      3h38m
aerocluster-source-xdr       ClusterIP   None             <none>        3000/TCP            33m
aerocluster-source-xdr-0-0   NodePort    10.100.116.173   <none>        3000:31999/TCP      33m
xdr-proxy                    ClusterIP   10.100.44.96     <none>        8901/TCP,8902/TCP   41m

Insert a source record using the following command in AQL

kubectl run -it --rm --restart=Never aerospike-tool -n aerospike --image=aerospike/aerospike-tools:latest -- aql -U admin -P admin123 -h aerocluster-source-xdr-0-0

insert into test.a1 (PK,a,b,c,d) values(1,"A","B","C","D")
OK, 1 record affected.

aql> select * from test
+----+-----+-----+-----+-----+
| PK | a   | b   | c   | d   |
+----+-----+-----+-----+-----+
| 1  | "A" | "B" | "C" | "D" |
+----+-----+-----+-----+-----+
1 row in set (0.023 secs)

Now, run the following select query in the destination cluster.

kubectl run -it --rm --restart=Never aerospike-tool -n aerospike --image=aerospike/aerospike-tools:latest -- aql -U admin -P admin123 -h aerocluster-dest-xdr-0-0

aql> select * from test
+----+-----+-----+-----+-----+
| PK | a   | b   | c   | d   |
+----+-----+-----+-----+-----+
| 1  | "A" | "B" | "C" | "D" |
+----+-----+-----+-----+-----+
1 row in set (0.031 secs)

OK

Interim summary

At this point, it's confirmed that xdr-proxy is doing exactly what It should do.

If if you review the log file for the initial two xdr-proxies scheduled, you should see userKey=1.

kubectl logs xdr-proxy-7d9fccd6c8-5q7gn -n aerospike | grep record-parser
2022-12-08 14:53:50.607 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=a1, digest=[120, 48, -23, -90, 110, 126, 84, -1, 114, -116, -9, -21, 28, 75, 126, -68, -51, 83, 31, -117], userKey=1, lastUpdateTimeMs=1670511230565, userKeyString=1, digestString='eDDppm5+VP9yjPfrHEt+vM1TH4s=')

kubectl logs xdr-proxy-7d9fccd6c8-f5zkt -n aerospike | grep \
record-parser
(none)

Scaling the XDR-Proxies

Go ahead and scale up the xdr-proxy to six pods by editing the file xdr-proxy-deployment.yaml and then apply the changes.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: xdr-proxy
  namespace: aerospike
  labels:
    app: xdr-proxy
spec:
  replicas: 6
  selector:
    matchLabels:
      app: xdr-proxy
...
kubectl apply -f xdr-proxy-deployment.yaml

You can achieve exactly the same by running the following command.

kubectl scale deploy xdr-proxy -n aerospike --replicas=6

You should now have six instances of the xdr-proxies running.

kubectl get po -n aerospike -w
NAME                         READY   STATUS    RESTARTS   AGE
aerocluster-dest-xdr-0-0     1/1     Running   0          3h50m
aerocluster-source-xdr-0-0   1/1     Running   0          75m
xdr-proxy-7d9fccd6c8-49ttl   1/1     Running   0          7s
xdr-proxy-7d9fccd6c8-5q7gn   1/1     Running   0          83m
xdr-proxy-7d9fccd6c8-c4j7k   1/1     Running   0          7s
xdr-proxy-7d9fccd6c8-f5zkt   1/1     Running   0          83m
xdr-proxy-7d9fccd6c8-lscbg   1/1     Running   0          7s
xdr-proxy-7d9fccd6c8-r56vs   1/1     Running   0          7s

Add some sample messages in the source cluster with primary keys 5, 6, and 7. Note that in Aerospike, a primary key serves as a unique identifier for a record within a specific set in a specific namespace. This key is a unique identifier for addressing the record for any operation.

aql> insert into test (PK,a,b) values (5,"A","B")
OK, 1 record affected.
aql> insert into test (PK,a,b) values (6,"A","B")
OK, 1 record affected.
aql> insert into test (PK,a,b) values (7,"A","B")
OK, 1 record affected.

Notice how userKey=5, userKey=6, and userKey=7 have been shipped across to the newly scaled xdr-proxies. Run the commands below to see the same,

kubectl logs xdr-proxy-7d9fccd6c8-5q7gn -n aerospike | grep \
record-parser
2022-12-08 14:53:50.607 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=a1, digest=[120, 48, -23, -90, 110, 126, 84, -1, 114, -116, -9, -21, 28, 75, 126, -68, -51, 83, 31, -117], userKey=1, lastUpdateTimeMs=1670511230565, userKeyString=1, digestString='eDDppm5+VP9yjPfrHEt+vM1TH4s=')

kubectl logs xdr-proxy-7d9fccd6c8-f5zkt -n aerospike | grep \
record-parser
(none)

kubectl logs xdr-proxy-7d9fccd6c8-49ttl -n aerospike | grep \
record-parser
(none)

kubectl logs xdr-proxy-7d9fccd6c8-c4j7k -n aerospike | grep \
record-parser
2022-12-08 15:05:37.511 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=null, digest=[-88, 104, 104, -114, 19, -44, -19, 29, -15, 18, 118, 72, -117, -106, -28, 21, -48, 50, 26, 113], userKey=7, lastUpdateTimeMs=1670511937250, userKeyString=7, digestString='qGhojhPU7R3xEnZIi5bkFdAyGnE=')

kubectl logs xdr-proxy-7d9fccd6c8-lscbg -n aerospike | grep record-parser
2022-12-08 15:05:27.548 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=null, digest=[68, 4, -94, -44, -75, 112, -102, 73, -120, 41, -101, -120, 33, -111, 15, -114, -85, 46, -2, 80], userKey=5, lastUpdateTimeMs=1670511927465, userKeyString=5, digestString='RASi1LVwmkmIKZuIIZEPjqsu/lA=')
2022-12-08 15:05:32.300 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=null, digest=[33, -100, 127, 120, 17, 45, -79, 115, -40, 53, -70, -57, 120, 73, 20, -50, -99, -98, -104, 85], userKey=6, lastUpdateTimeMs=1670511932288, userKeyString=6, digestString='IZx/eBEtsXPYNbrHeEkUzp2emFU=')

kubectl logs xdr-proxy-7d9fccd6c8-r56vs -n aerospike | grep record-parser
(none)

However, we need to consider one critical factor. When data is actively flowing between source and destination clusters, the existing cached list of xdr-proxy connections from the source cluster will not be refreshed just because you added a bunch of new xdr-proxies. Consequently, the newly scaled xdr-proxies you have just scheduled will not be utilized immediately.

To demonstrate this, let's add data to the source cluster using Aerospike's benchmark tool. At the same time, we will scale the xdr-proxies on the destination side and observe the results.

Add data to the source cluster

Before you begin, reduce the xdr-proxy server count to one (1) to clarify the observations. In this example, I use my local machine, which has the benchmark tool installed, to send data to the source EC2 instances. To do this, you’ll need to obtain the external IP address of the source cluster’s Kubernetes service. You can download the benchmark tool from https://aerospike.com/docs/tools/install.

kubectl get AerospikeCluster aerocluster-source-xdr -n aerospike  -o yaml

...
pods:
    aerocluster-source-xdr-0-0:
      aerospike:
        accessEndpoints:
        - 192.168.41.63:31999
        alternateAccessEndpoints:
        - 54.173.138.131:31999
        clusterName: aerocluster-source-xdr
        nodeID: 0a0
        tlsAccessEndpoints: []
        tlsAlternateAccessEndpoints: []
        tlsName: ""
      aerospikeConfigHash: 4aacb9809beaa01d99a9f00293c9f7dc141845f8
      hostExternalIP: 54.173.138.131
      hostInternalIP: 192.168.41.63
      image: aerospike/aerospike-server-enterprise:6.1.0.2
      initializedVolumes:
      - workdir
      - ns
      networkPolicyHash: acbbfab3668e1fceeed201139d1173f00095667e
      podIP: 192.168.50.203
      podPort: 3000
      podSpecHash: 972dc2a779fe9ab407212b547d54d3a72ecef259
      servicePort: 31999
...

You may need to add a firewall rule to allow traffic into the Kubernetes service. Connect the asbenchmark tool to start writing traffic using the public IP address for the NodePort Service.

asbenchmark -h 54.173.138.131:31999 -Uadmin -Padmin123 -z 10 -servicesAlternate -w RU,0 -o B256

Scale up the xdr-proxy servers from one to two and check the logs of both proxies to see what messages have been received. In a production environment, you should always disable the unnecessary logging.

Notice how no data has passed through the new xdr-proxy instance xdr-proxy-7d9fccd6c8-s2tzt.

kubectl -n aerospike logs xdr-proxy-7d9fccd6c8-s2tzt | grep \
record-parser
(none)

kubectl -n aerospike logs xdr-proxy-7d9fccd6c8-5q7gn | grep \
record-parser
...
2022-12-08 18:21:36.158 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[51, -120, 114, -17, -44, 72, 123, 125, 50, 92, 3, 110, -21, -38, 74, 25, 42, 35, 117, 72], userKey=null, lastUpdateTimeMs=1670523696059, userKeyString=null, digestString='M4hy79RIe30yXANu69pKGSojdUg=')
2022-12-08 18:21:36.158 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[124, -6, -70, -23, 44, 41, -19, 40, -11, -16, 126, 120, 81, -113, -112, -79, 66, 77, -99, -6], userKey=null, lastUpdateTimeMs=1670523696059, userKeyString=null, digestString='fPq66Swp7Sj18H54UY+QsUJNnfo=')
2022-12-08 18:21:36.158 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-127, -118, 63, -32, -74, 60, -60, 86, 31, -119, -1, -105, -108, -59, 111, 48, -34, -61, -108, -5], userKey=null, lastUpdateTimeMs=1670523696105, userKeyString=null, digestString='gYo/4LY8xFYfif+XlMVvMN7DlPs=')
2022-12-08 18:21:36.256 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[99, 1, 80, 2, 76, -43, 125, 77, 47, 8, 6, 35, 49, 117, -35, 54, 120, -29, 118, -72], userKey=null, lastUpdateTimeMs=1670523696178, userKeyString=null, digestString='YwFQAkzVfU0vCAYjMXXdNnjjdrg=')
2022-12-08 18:21:36.257 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-88, -46, -48, -33, 77, -120, 123, -101, -70, -20, -96, -104, -51, -90, 28, -15, 70, 11, 118, 83], userKey=null, lastUpdateTimeMs=1670523696202, userKeyString=null, digestString='qNLQ302Ie5u67KCYzaYc8UYLdlM=')
2022-12-08 18:21:36.257 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-97, 18, 28, -43, 75, 42, -22, -126, -61, -108, -36, 118, -86, -105, -52, 119, -39, -33, -127, -76], userKey=null, lastUpdateTimeMs=1670523696175, userKeyString=null, digestString='nxIc1Usq6oLDlNx2qpfMd9nfgbQ=')
2022-12-08 18:21:36.257 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-120, 103, -51, 57, -71, -106, 13, -48, 100, 28, 59, -3, -39, -56, -67, -103, 29, 36, 75, 119], userKey=null, lastUpdateTimeMs=1670523696191, userKeyString=null, digestString='iGfNObmWDdBkHDv92ci9mR0kS3c=')
2022-12-08 18:21:36.257 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-47, 88, -13, 13, -35, 77, 24, 22, -40, -61, -118, -115, 82, 13, 127, -125, 53, 66, -22, -8], userKey=null, lastUpdateTimeMs=1670523696233, userKeyString=null, digestString='0VjzDd1NGBbYw4qNUg1/gzVC6vg=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-63, -11, 93, -90, 47, 29, -63, 36, 12, 53, -86, 84, 57, -125, 16, 43, -18, 93, 56, 9], userKey=null, lastUpdateTimeMs=1670523696186, userKeyString=null, digestString='wfVdpi8dwSQMNapUOYMQK+5dOAk=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-23, 101, -114, -87, -52, 107, 36, 113, 101, 33, -16, 82, -95, 97, 34, -121, 82, -97, 40, 59], userKey=null, lastUpdateTimeMs=1670523696145, userKeyString=null, digestString='6WWOqcxrJHFlIfBSoWEih1KfKDs=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-77, -118, 49, 4, -75, 123, 81, 2, -103, -73, 42, -70, -54, 95, 98, 23, 73, 66, -86, 7], userKey=null, lastUpdateTimeMs=1670523696230, userKeyString=null, digestString='s4oxBLV7UQKZtyq6yl9iF0lCqgc=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[50, -38, -31, -54, -122, -113, -38, 88, 15, 7, 96, 51, -92, -25, 60, -104, 26, 113, -117, -82], userKey=null, lastUpdateTimeMs=1670523696157, userKeyString=null, digestString='MtrhyoaP2lgPB2AzpOc8mBpxi64=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[-77, 31, 67, -18, -52, -114, 42, -18, 36, -111, 89, 62, 109, 114, -54, 54, -121, -110, -88, -108], userKey=null, lastUpdateTimeMs=1670523696206, userKeyString=null, digestString='sx9D7syOKu4kkVk+bXLKNoeSqJQ=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[73, 11, 98, -50, 32, 12, 0, -50, 22, -101, -108, 18, 38, 7, -65, 6, -58, 60, -6, -33], userKey=null, lastUpdateTimeMs=1670523696171, userKeyString=null, digestString='SQtiziAMAM4Wm5QSJge/BsY8+t8=')
2022-12-08 18:21:36.258 GMT DEBUG record-parser - parsed message fields: key=Key(namespace='test', set=testset, digest=[109, -82, 24, 53, 35, 89, -72, -117, -22, 79, 119, -89, 56, -5, 0, -103, -54, 51, 25, 126], userKey=null, lastUpdateTimeMs=1670523696146, userKeyString=null, digestString='ba4YNSNZuIvqT3enOPsAmcozGX4=')

Achieve dependable resiliency with Aerospike

Aerospike's XDR feature provides a reliable solution for mitigating the risk of cluster unavailability. By asynchronously replicating data between data centers, users can ensure availability. This step-by-step walkthrough demonstrates how to accomplish this seamlessly in a Kubernetes environment using the xdr-proxy. With the Aerospike Kubernetes Operator, you can effortlessly avoid network complications and achieve optimal performance with minimal effort.

Share your experience! Your feedback is important to us. Join our Aerospike community!

Using auto-values in Aerospike LINQPad driver

Richard Andersen — Thu, 08 Feb 2024 20:03:53 +0000

When working with NoSQL databases like Aerospike, which are designed for unstructured data storage, adopting strongly typed languages like C# may introduce programming complexities. These can stem from verifying the presence of "bins" (akin to columns) within a record and bin data types that can change between records.

Auto-Values (AValue is the class) simplify the need to check, cast, or convert values between Aerospike data types and .NET data types. They provide a rich set of operations and seamlessly work with any type of data, including collections and JSON. They provide protection against null reference exceptions, invalid cast exceptions, conversion exceptions, etc.
To explore the LINQPad or the Aerospike LINQPad driver in more detail, kindly click on the respective links provided.

Example set

The following illustrates an Aerospike set where different records have a varied number of bins, and bins in the record set can have varying data types (e.g., bin BinA can be a double, long, string, or list in different records).

Display of the DemoTypes set

Note that the LINQPad connection tree (left side) is expanded, illustrating the bins for the set DemoTypes. Under the set name, each bin is listed, displaying its associated data type. Take note of the symbols after the data type. A * indicates this bin has different data types between records. A ? indicates this bin wasn’t present in some of the records within this set. These symbols provide a quick view of the set and expected values.
In the Results pane (bottom-right), bins with null are not present in that record.

Simple cast examples

Below are examples that illustrate the difference between using Aerospike data types, C# data types, and Auto-Values.
Let’s begin with an example where Auto-Values are not used:

Shows that casting to the proper data type is required

Notice that there are no results from the query, even though it should have matched one record. The reason it failed is because “456” is an Int32, and the DB numeric values are Int64. So, “456” needs to be cast to Int64 (long). Also, we need to check to make sure the bin is present in all records; otherwise, a null exception is thrown.

Below is an example showing these changes:

Updated query using required cast

Let’s look at the same query using Auto-Values:

Using Auto-Values doesn’t require testing or casting

In the given example, you do not have to verify the existence of the bin or explicitly cast "456" to a long data type.
Auto-Values take care of not only casting but also converting. Conversion between any of the .NET primitive types to or from any Aerospike data types is seamless. It also handles nullable values, JSON objects, and collection data types (CDT).

Auto-Values support all standard operations like ==, >, <, etc. They also support all the equality and comparison operations.

Below are some additional examples:

Queries above show different uses of Auto-Values

Handling unsupported data types

Aerospike supports a limited set of data types. They include:

String
Integer (Int64)
Double
Boolean
Blob/Bytes (byte[])
List
Map (Directory)
Geospatial (GeoJSON.NET)

The driver has extended support by allowing the following .NET data types:

DateTime, DateTimeOffset, TimeSpan: These values are mapped to either an Aerospike string or long data types depending on the connection or API configuration. Auto-Values will automatically take care of these conversions to or from the database.
JSON: These values are mapped to an Aerospike Map data type where the JSON property name is the key in the Map, and the JSON value is the Map value. JSON arrays are mapped to Aerospike “List.” Nested JSON structures are completely supported, and the driver takes care of all conversions to or from the database.

Below are examples where .NET DateTimeOffset, DateTime, and TimeSpan instances are used to obtain matching records. The values stored in the Aerospike database are either strings or an Int64 value (number of nanoseconds from Unix Epoch). The driver will manage the conversion between these different types.

Queries using .NET DateTime, DateTimeOffset, and Timespan

By the way, you can control how values are converted and displayed through the properties located in the "Display/Conversion Options" section of the connection dialog.

Conversion

Auto-Values provide a rich set of conversion functions, including:

Convert<T> - Will try to convert to the provided .NET data type. If it cannot, an invalid cast exception is thrown.
Is{data type} – Use to test if the underlying .NET data type is that {date type}. Examples of these functions are: IsList, IsCDT, IsJSON, IsInt16, IsNumeric, IsString, etc.
To{data type} – Will try to convert to the .NET {data type}, if possible. Note {data type} can be Dictionary, List, .NET native type, etc. Examples of these functions are: ToList, ToBoolean, ToByte, etc.
TryGetValue<T> – This will try to match a provided value and convert the value into the provided .NET data type. If not successful, the default value of that data type and/or false is returned, depending on usage. This function can be applied against CDTs or non-CDT values.
TryGetValue – This will try to match the provided value. If successful, the matched value is returned as an Auto-Value. If not, an empty Auto-Value, false, or null can be returned depending on usage. This function can be applied against CDTs or non-CDT values.
AsEnumerable – This will convert an Auto-Value into an enumerable object if it is a database CDT. If not a CDT, an empty enumeration is returned. All elements in the CDT are scanned and converted into Auto-Values. This will provide the highest level of protection against invalid casts or null value reference exceptions. This also allows for using the advanced Auto-Value functions outlined in the Collection Data Types section.
Implicit Casting – Auto-Values know how to implicitly cast from an Aerospike data type to any .NET primary type without explicitly providing the type. Detailed documentation can be found through IntelliSense or by reviewing the function’s respective documentation.

Collection data types

Auto-Values can seamlessly be used to find elements within CDTs. These functions are based on standard C# methods. Examples of these operations are:

Contains – Returns true if the matching value is contained in a bin’s value or an element within a CDT. The matching options determine how the matches occur.
FindAll – Returns a collection of matching Auto-Values. A match can occur as a bin’s value or an element within a CDT. The matching options determine how the matches occur. If no matches are found, an empty enumerable is returned.
TryGetValue – Returns the first matching value (as an Auto-Value) contained in a bin’s value or an element within a CDT. If the Auto-Value is not found, an Empty Auto-Value is returned.
OfType<T> - Tries to cast the Auto-Value to the provided .NET type, creating a new collection of those types. If a value cannot be cast, it will be ignored.
Cast<T> -- Will cast the Auto-Value to the provided .NET type. If it cannot be cast, an invalid cast exception will occur.
Convert<T> - Will try to convert the Auto-Value, resulting in a collection of converted .NET values. If an Auto-Value cannot be converted, it will be ignored.

Below are some examples using “Contains” and “FindAll” methods:

Queries using the Contains and FindAll methods

You can find additional examples in the LINQPad sample folder.
Below is an example of “drilling” into several different sub-collections using LINQ. This example finds all invoices for a certain song track, resulting in a collection of customers. This example is located in the CDT-Json-Docs sample LINQPad script.

LINQ Query

Primary key

Auto-Values also extend to primary keys. They have the same features as found in the Bin Auto-Values. There are a few additional features, such as digest values.
If the primary key value is not saved (only the digest is used), matching to a value can still occur using Auto-Values.
Below is an example where the digest is only available; note that the actual value must be used.

Query using a string value by means of the database digest

LINQPad driver samples

You can find samples for the Aerospike LINQPad driver in the samples folder or under the “Samples” tab in LINQPad.

Using AWS IAM for client authentication

Naresh Maharaj — Wed, 20 Dec 2023 20:11:36 +0000

In this blog post, we detail how to create an Amazon Managed Streaming for Apache Kafka (Amazon MSK) resource using AWS Identity and Access Management (AWS IAM) in roles and policies to authenticate user access. In the initial step, we establish an Aerospike Database cluster and insert sample messages into the database. Subsequently, we observe in real time how these messages are streamed to Amazon MSK using Aerospike's Kafka Source Connector. Below we provide a comprehensive, step-by-step guide for users to successfully implement this process.

AWS MKS Kafka

In this section, you will set up a simple three-node Kafka cluster.

Visit the AWS console and select MSK service.

Create a new cluster by selecting Create Cluster → Quick Create.

Select the provisioned cluster and instance type of kafka.t3.small.

Select the EBS storage type per broker of 10 GB.

NOTE: Take note of the VPC, subnets, and security group ID, as you will require these details later in the article.

The next step is the critical step where you will create the AWS IAM policy and roles. This setup ensures that the Aerospike Database authenticates using AWS IAM to write data to MSK.

From the AWS Console, select the AWS IAM service.

To create a new AWS IAM policy, copy the following JSON and paste it in the JSON tab. Replace region:Account-ID with your own region and AWS account ID.

Save the policy and name it msk-tutorial-policy.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:AlterCluster",
                "kafka-cluster:DescribeCluster"
            ],
            "Resource": [
                "arn:aws:kafka:region:Account-ID:cluster/MSKTutorialCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:*Topic*",
                "kafka-cluster:WriteData",
                "kafka-cluster:ReadData"
            ],
            "Resource": [
                "arn:aws:kafka:region:Account-ID:topic/MSKTutorialCluster/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": [
                "arn:aws:kafka:region:Account-ID:group/MSKTutorialCluster/*"
            ]
        }
    ]
}

Create the IAM role.

Under Common Use Cases, select EC2 and then Next.

Under Permissions, select the policy named msk-tutorial-policy and then Next.

Give the role a name like msk-tutorial-role and click the Create Role button.

Kafka client machine

Next, create a client machine to install the Kafka tools necessary to access our MSK cluster.

Create a new ec2 instance using type t2.micro

Use the default AMI: Amazon Linux 2023

The AMI may be different depending on your region

Create a key-pair if required. I am using an already existing key-pair.

Under Advanced Options.IAM instance profile, select the IAM role created earlier.

Launch the instance!
Under instances launched, choose the instance you just created. Click on the ‘Security’ tab and note the security group associated with this instance.
e.g., sg-0914e6271c97ae4c9 (launch-wizard-1)
Navigate to the VPC section and select Security Groups from the left-hand menu. Locate the security group associated with the MSK cluster, such as sg-e5f51dfb, and choose Edit Inbound Rules.
Create a new rule to allow all traffic from the new ec2 instance.

Kafka topics

After successfully establishing your initial Kafka cluster and Kafka client machine, proceed to conduct testing. Verify the functionality by accessing the MSK cluster, creating a topic, producing and consuming sample messages, and ensuring that everything operates as anticipated.

From the MSK Cluster, note the Kafka version being used. This examples uses 2.8.1.
From the Kafka client machine, install Java 11+.

sudo yum -y install java-11

Download Apache Kafka using wget, then extract the archive using tar.

wget https://archive.apache.org/dist/kafka/2.8.1/kafka_2.12-2.8.1.tgz
tar -xzf kafka_2.12-2.8.1.tgz

To use IAM, you will need the MSK IAM Auth jar file. Download the jar to the Kafka libs folder you just extracted.

cd kafka_2.12-2.8.1/libs/
wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar
cd ../bin/

Create a file called client.properties to use when authenticating to MSK. It will define the SASL mechanism to use and reference the Java class file that will handle your IAM callbacks.

cat <<EOF> client.properties
security.protocol=SASL_SSL
sasl.mechanism=AWS_MSK_IAM
sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;
sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
EOF

Creating topics

Go to the AWS Console and view the MSK Cluster Client Information. There will be three endpoints to choose from, but you only require one.

Example choose:

B-2.msktutorialcluster.450050.c11.kafka.us-east-1.amazonaws.com:9098

From the kafka/bin folder, run the command to create a topic. Let's call it aerospike-airforce-1.

export BootstrapServerString="b-2.msktutorialcluster.450050.c11.kafka.us-east-1.amazonaws.com:9098"

./kafka-topics.sh --create --bootstrap-server $BootstrapServerString --command-config client.properties --replication-factor 3 --partitions 1 --topic aerospike-airforce-1

Listing topics

To list the topics, use the following command. Notice our latest topic, called aerospike-airforce-1, just showed up.

./kafka-topics.sh --bootstrap-server $BootstrapServerString --command-config client.properties --list

MSKTutorialTopic
__amazon_msk_canary
__consumer_offsets
aerospike
aerospike-airforce-1

Producer and consumer

I agree that this is more of a Kafka-101 rather than a straightforward Hello-World scenario. Nonetheless, it is essential to test our configuration by sending and receiving messages from the designated Kafka topic before proceeding further.

Produce some messages by opening a new window and running the following Kafka producer command. Type three or four messages, hitting the 'Return' key after each message

./kafka-console-producer.sh --broker-list $BootstrapServerString --producer.config client.properties --topic aerospike-airforce-1
>Instrument Check
>Pre flight checks confirmed
>Ready for takeoff
>Full throttle, flaps

You're now ready to start a client consumer application. Open a new window and run the consumer. You should now see the same messages you published earlier.

./kafka-console-consumer.sh --bootstrap-server $BootstrapServerString --consumer.config client.properties --topic aerospike-airforce-1 --from-beginning
Instrument Check
Pre flight checks confirmed
Ready for takeoff
Full throttle, flaps

Database source

Let's review your achievements thus far. You've established a 3-node Kafka cluster in AWS utilizing MSK, incorporating IAM roles and permissions. Additionally, you have successfully created topics and demonstrated the production and consumption of messages using the IAM credentials established during the setup.

The next phase of your journey involves installing the Aerospike Database, inserting messages, and configuring a simple XDR component. XDR is a Cross Datacenter Replication tool and is crucial for transmitting data from the Aerospike Database to the Aerospike Kafka Source Connector allowing us to subsequently forward messages to Amazon MSK.

Create the Aerospike Database

Start by creating a new ec2 instance. For this demo, you can use Linux Centos 8

Rocky 8 AMI: ami-043ceee68871e0bb5 ( us-east-1 )

Select the instance type as t2.medium.

Add the extra volume for the Aerospike data storage layer. EBS volume is all that is required for now.

Launch the instance and connect to the host using ssh. If you have an Aerospike license feature file, upload it to the instance.

Install the Aerospike Database server

Run the following to install the Aerospike Database Server.

export VER="6.1.0.2"
sudo yum install java python3 openssl-devel wget git gcc maven bind-utils sysstat nc -y
wget -O aerospike-tools.tgz 'https://www.aerospike.com/download/tools/latest/artifact/el8'
tar -xvf aerospike-tools.tgz
cd aerospike-tools_*
sudo ./dep-check
sudo ./asinstall
wget -O aerospike.tgz https://enterprise.aerospike.com/enterprise/download/server/$VER/artifact/el8
tar -xvf aerospike.tgz
cd aerospike-server-enterprise-$VER-el8
sudo ./asinstall
sudo mkdir -p /var/log/aerospike/
sudo systemctl enable aerospike

Confirm the storage disk for Aerospike.

lsblk
NAME    MAJ:MIN RM SIZE RO TYPE MOUNTPOINT
xvda    202:0    0  10G  0 disk
└─xvda1 202:1    0  10G  0 part /
xvdb    202:16   0  10G  0 disk   <<----------------- This one!

When its data is available, replace the Aerospike configuration file under /etc/aerospike/aerospike.conf with the configuration file listed below, also replacing the following lines:
- Under heartbeat.address add in your internal 172.x.x.x address
- For xdr.dc.node-address-port enter the {kafka-client-machine-address}:8080

Aerospike Database configuration file for use with `systemd`

service {
  # paxos-single-replica-limit 1 # Number of nodes where the replica count is automatically
  proto-fd-max 15000
  service-threads 10
  feature-key-file /etc/aerospike/features.conf
  node-id A1
  cluster-name CLA
}

logging {
  file /var/log/aerospike/aerospike.log {
    context any info
  }
}

# public and private addresses
network {
  service {
    address any
    port 3000
  }

  heartbeat {
    mode mesh
    address 172.31.94.201
    port 3002 # Heartbeat port for this node.
    interval 150 # controls how often to send a heartbeat packet
    timeout 10 # number of intervals after which a node is considered to be missing
  }

  fabric {
    port 3001
  }

  info {
    port 3003
  }
}

namespace test {
  replication-factor 2
  memory-size 40G
  default-ttl 0
  index-type shmem
  high-water-disk-pct 50
  high-water-memory-pct 60
  stop-writes-pct 90
  nsup-period 0

  storage-engine device {
    device /dev/xvdb
    data-in-memory false
    write-block-size 128K
    min-avail-pct 5
  }
}

xdr {
  # Change notification XDR block that round-robins between two connector nodes
  dc aerospike-kafka-source {
    connector true
    node-address-port 172.31.58.190 8080
    namespace test {
    }
  }
}

Start the Aerospike service

Copy the license feature file to the aerospike configuration directory.

sudo cp features.conf /etc/aerospike/

Start the Aerospike server and check the logs to ensure there are no errors.

sudo systemctl start aerospike

sudo systemctl status aerospike

Aerospike Kafka Source Connector

The seamless flow of data from Aerospike Database Enterprise Edition to Apache Kafka hinges on the utilization of the Aerospike Kafka source (outbound) connector. This connector subscribes to change notifications. Upon receiving these notifications, the connector converts them into messages, which are dispatched to Kafka topics. Going back to the ec2 instance you created earlier with our Kafka client configured, go ahead and install the Aerospike Kafka Source Connector. This is your outbound connector to send data from the Aerospike to MSK.

sudo yum install java #( install 11+ JDK )

wget https://enterprise.aerospike.com/artifacts/enterprise/aerospike-kafka-outbound/5.0.1/aerospike-kafka-outbound-5.0.1-1.noarch.rpm

sudo rpm -i aerospike-kafka-outbound-5.0.0-1.noarch.rpm

Configure the outbound connector

The terms “outbound” and “source connector” are used interchangeably in this article.

Locate the following file on the Kafka client box: /etc/aerospike-kafka-outbound/aerospike-kafka-outbound.yml.
Replace the broker address for one of the node addresses in the MSK Kafka cluster producer-props.bootstrap.servers.
Then add the following contents to the file with the changes that have been outlined.

# Change the configuration for your use case.
#
# Refer to https://www.aerospike.com/docs/connectors/enterprise/kafka/outbound/configuration/index.html
# for details.

# The connector's listening ports, TLS, and network interface.
service:
  port: 8080

# Format of the Kafka destination message.
format:
  mode: flat-json
  metadata-key: metadata

# Aerospike record routing to a Kafka destination.
routing:
  mode: static
  destination: aerospike

# Kafka producer initialization properties.
producer-props:
  bootstrap.servers:
    - b-3.msktutorialcluster.450050.c11.kafka.us-east-1.amazonaws.com:9098
  ssl.truststore.location: /etc/aerospike-kafka-outbound/kafka.client.truststore.jks
  ssl.truststore.password: changeit
  security.protocol: SASL_SSL
  sasl.mechanism: AWS_MSK_IAM
  sasl.jaas.config: software.amazon.msk.auth.iam.IAMLoginModule required awsProfileName=default;
  sasl.client.callback.handler.class: software.amazon.msk.auth.iam.IAMClientCallbackHandler

# The logging properties.
logging:
  file: /var/log/aerospike-kafka-outbound/aerospike-kafka-outbound.log
  enable-console-logging: true
  levels:
    root: debug
    record-parser: debug
    server: debug
    com.aerospike.connect: debug
  ticker-interval: 3600

Create the CA certificate trust store for use in the Kafka Outbound Connector config. You can see the SSL trust store location referenced in the file above as ssl.truststore.location.

sudo cp /usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts /etc/aerospike-kafka-outbound/kafka.client.truststore.jks

sudo chmod 755 /etc/aerospike-kafka-outbound/kafka.client.truststore.jks

Finally, make the AWS IAM Kafka Auth Jar file available to the Aerospike Outbound Kafka Connector. This is the same jar file that you downloaded and added to the kafka/libs folder.

sudo cp kafka_2.12-2.8.1/libs/aws-msk-iam-auth-1.1.1-all.jar /opt/aerospike-kafka-outbound/lib/aws-msk-iam-auth-1.1.1-all.jar

Start the service.

sudo systemctl enable aerospike-kafka-outbound

sudo systemctl start aerospike-kafka-outbound

Send data from Aerospike to Kafka

Open a separate window so you can list all messages on the Aerospike Kafka topic. Start by adding one of the private endpoint bootstrap servers as an environment variable for ease of use.

export BootstrapServerString="b-3.msktutorialcluster.450050.c11.kafka.us-east-1.amazonaws.com:9098"

Run the consumer client as follows:

./kafka-console-consumer.sh --bootstrap-server $BootstrapServerString --consumer.config client.properties --topic aerospike --from-beginning

In a new window, start AQL, the Aerospike command line client which connects to your Aerospike Database.

aql -U auser -P a-secret-pwd

Insert some data

insert into test (pk, a) values(400, "Your winning lottery ticket awaits you")

Check to see if the message appears in the Kafka consumer window

{"metadata":{"namespace":"test","userKey":400,"digest":"W7eGav2hKfOU00xx7mnOPYa2uCo=","msg":"write","gen":1,"lut":1681488437767,"exp":0},"a":"Your winning lottery ticket awaits you"}

Conclusion

You've just discovered how straightforward it is to transmit data from Aerospike to AWS MSK Kafka while ensuring client authentication through AWS IAM permissions! From establishing an Aerospike Database from scratch to configuring the AWS MSK Kafka cluster and employing the Aerospike Outbound Kafka Connector, you've effortlessly constructed a real-time streaming data pipeline. Congratulations on this accomplishment!

Share your experience! Your feedback is important to us. Join our Aerospike community!

A developer’s introduction to graph databases

Alex Infanzon — Thu, 30 Nov 2023 02:31:18 +0000

Get ready to deploy a real-world dataset on Aerospike Graph and execute queries in just a few minutes.

Ready to dive into new realms of graph technology and discover fascinating insights with each click? Whether you're a curious beginner or a seasoned graph explorer searching for fresh perspectives, this blog is meant for you.

Aerospike Graph at a glance

Aerospike Graph leverages Apache TinkerPop, an open-source graph computing framework, and Gremlin, a graph traversal language. In other words, Gremlin is a query language, like SQL, for traversing and manipulating graph-structured data. Using Gremlin, you can explore specific graph traversal sequences known as traversals. Our post, Aerospike and gdotv partnership delivers rich graph data visualization goes further into the capabilities at your disposal when using a Gremlin IDE.

Aerospike Graph is built on top of the Aerospike Database, a high-performance, scalable, and reliable database engine. Aerospike Graph can build and operate large-scale graph applications across many use cases, including customer 360 and fraud prevention, among others.

Working with graph data

In this example, we examine various queries that explore a graph which contains a model of the worldwide air route network. The air-routes graph schema (see figure below) models a transportation network for air routes. It represents airports, countries, and continents as vertices and flight routes (i.e., routes) and continents or countries (i.e., the airport's location) as edges between the vertices. Each vertex (e.g., airport) has properties such as an airport code, name, location (latitude and longitude), and time zone. Similarly, edges (routes) can have properties such as distance between airports.

The schema includes the following elements:

Airport vertex: Represents an airport in the transportation network. It has properties such as:
- code: The unique code assigned to the airport.
- name: The name of the airport.
- location: The geographical coordinates of the airport (latitude and longitude).
- timezone: The time zone of the airport.
Route edge: Represents a flight route between two airports. It has properties such as:
- airline: The name of the airline operating the flight.
- distance: The distance between the airports.

Additionally, the schema could include other optional properties or additional vertex and edge types to represent more complex relationships, such as intermediate stops or connecting flights. The specific implementation of the graph schema may vary depending on the use case and your requirements. The remainder of the blog describes the software prerequisites, the Docker images you need, how to connect and load the data, and how to execute some queries.

Why Docker containers? If you are a developer working on multiple machines, each time you switch a machine, you need to set up and configure the database separately.

Using Aerospike Database and Aerospike Graph inside a Docker container, you can quickly spin up a sandbox environment and focus on actual development rather than infrastructure setup. The same is true for the production environment.

1. Software prerequisites

As usual, there are a few prerequisites that need to be taken care of before you begin. Following is a list of the software used to run this JupyterLab Notebook.

The next section provides the steps to download the Docker images and launch the Graph Service and Database containers.

Download Aerospike software

Once you have Docker installed, the simplest way to get up and running is to pull the Aerospike images from the Docker hub repository. In a terminal window, execute the following pull commands:

docker pull aerospike/aerospike-serverdocker
docker pull aerospike/aerospike-graph-servicedocker

You’ll also need a trial license key for the Aerospike Graph Service. Get started by downloading a free trial.

If you would like detailed steps on how to start the containers, take a look at the Aerospike Graph Using Docker documentation. However, the steps below should be enough to get you started.

Set an environment variable with the license key (aka, Feature Key).

export FEATKEY=$(base64 -i /<YOUR_PATH>/<YOUR_FEATURE_KEY.conf>)

For example:

export FEATKEY=$(base64 -i /Users/Shared/setup/as_featurekey.conf)

Launch the database container.

docker run -d -e "FEATURES=$FEATKEY" -e "FEATURE_KEY_FILE=env-b64:FEATURES" --name as_database -p 3000-3002:3000-3002 aerospike/aerospike-server-enterprise

Launch the graph service.

docker run -p8182:8182 --name as_graph -v <PATH_TO_YOUR_SHARED_DIR>:/opt/aerospike/etc/sampledata -e aerospike.client.namespace="test" -e aerospike.client.host="172.17.0.2:3000" -e aerospike.graph.index.vertex.label.enabled=true aerospike/aerospike-graph-service

For example:

docker run -p8182:8182 --name as_graph -v /Users/Shared/data/docker-bulk-load:/opt/aerospike/etc/sampledata -e aerospike.client.namespace="test" -e aerospike.client.host="172.17.0.2:3000" -e aerospike.graph.index.vertex.label.enabled=true aerospike/aerospike-graph-service

NOTE: The -v option allows you to bind a directory in your host operating system to a directory inside the Docker container. Make sure the air-routes-latest.graphml has been copied to the shared directory <PATH_TO_YOUR_SHARED_DIR> in the command above. You can download the air-routes-latest.graphml file form here.

Importing Python modules

Execute the following cell to import the packages you need. You might need to install them first using the pip command.

from importlib.metadata import version
from pandas.plotting import table
from ipywidgets import interact
from IPython.display import Markdown as md, display, HTML
from IPython.display import display_markdown
from gremlin_python.process.traversal import IO
from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
from gremlin_python.process.graph_traversal import GraphTraversalSource, __
from gremlin_python.process.traversal import Barrier, Bindings, Cardinality, Column, Direction, Operator, Order, P, Pop, Scope, T, WithOptions

import matplotlib.pyplot as plt
import nest_asyncio
import networkx as nx
import ipycytoscape
import jugri
import pandas as pd
import re

nest_asyncio.apply()

2. Connect to the graph service and load the air-route data

To traverse the vertices and edges in the graph, you must spawn a traversal object:

g = traversal().withRemote(DriverRemoteConnection('ws://0.0.0.0:8182/gremlin','g'))

The above Gremlin query represents a simple pattern for creating a traversal object using a remote connection. Here is the breakdown of each part:

g: This variable represents a reference to the traversal object. It allows you to build and execute queries on the graph database.
traversal(): This function creates a new traversal object. It is the entry point for executing queries in Gremlin.
withRemote(DriverRemoteConnection('ws://0.0.0.0:8182/gremlin','g')): This command configures the traversal object to use a remote connection to communicate with the Aerospike database. The DriverRemoteConnection class is used to establish the connection.
- ws://0.0.0.0:8182/gremlin: This parameter specifies the WebSocket endpoint URL of the server hosting the graph database.
- g: This parameter represents the specific traversal source on the server.

By executing this query, you can perform various graph database operations using the 'g' traversal object linked to the specified remote connection.

The next step is to populate the graph. The air-routes data should already be in the Aerospike Graph Service container. The file name is air-routes-latest.graphml and should be in the /opt/aerospike/etc/sampledata directory.

The file is in GraphML format. GraphML is an XML-based file format for graphs shown in the table below. On the left column, there is one entry (a record) for each one of the vertices in the graph schema, namely continent, airport, and county. The column on the right shows the format for edges.

Populate the graph (vertices and edges)

The following cell executes two statements. The first one drops all data in the graph. The second one loads the data using the io() step. The evaluationTimeout parameter prevents the loading operation from timing out when running for extended periods. In the following example, the timeout is set to 5 minutes in milliseconds (even though it takes a couple of seconds to load the air-routes data). You can adjust it as necessary.

g.V().drop().iterate()

g.with_("evaluationTimeout", 5 * 60 * 1000)\
  .io("/opt/aerospike/etc/sampledata/air-routes-latest.graphml")\
  .with_(IO.reader, IO.graphml)\
  .read()\
  .toList()

Once the data is loaded you can get a count of vertices and edges. To get a count of the total number of vertices loaded, execute the following statement:

g.V()
  .label()
    .groupCount()
  .toList()

g.V(): This step retrieves all the vertices in the graph.
.label(): This step extracts the label of each vertex.
.groupCount(): This step counts the occurrences of each unique label and creates a map with the label as the key and the count as a value.
.toList(): This step converts the map into a list, which can be returned as the final result of the Gremlin query.

NOTE: I am using Pandas data frames to capture the output of the queries and display the result either in tabular form or as a matplotlib chart.

v_cnt = pd.DataFrame.from_dict(g.V().label().groupCount().toList()).T   # .T to Transpose the dataframe
v_cnt.reset_index(inplace=True)
v_cnt.columns = ['vertex name','count']

To get the number of edges loaded, replace the g.V() step for g.E().

g.E()
  .label()
    .groupCount()
  .toList()

e_cnt = pd.DataFrame.from_dict(g.E().label().groupCount().toList()).T   # .T to Transpose the dataframe
e_cnt.reset_index(inplace=True)
e_cnt.columns = ['edge name','count']

The cell below uses matplotlib to display the counts as a bar chart.

display_markdown(f'''## Total Number of Vertices: {v_cnt['count'].sum()}; Total Number of Edges: {e_cnt['count'].sum()}''', raw=True)
fig, axes = plt.subplots(nrows=1, ncols=2, figsize=(20,5))
ax1 = v_cnt.sort_values(by = 'count', ascending=False,).plot.bar(ax=axes[0], title='Vertex Counts', x='vertex name', rot=50, color = 'red')
ax1.set_ylabel("vertices Count")
ax1.bar_label(ax1.containers[0])
ax2 = e_cnt.sort_values(by = 'count', ascending=False,).plot.bar(ax=axes[1], title='Edge Counts',x='edge name', rot=50, color = 'green')
ax2.set_ylabel("edges Count")
ax2.bar_label(ax2.containers[0])
plt.show()

After executing the above code you will get a count of vertices and edges in the graph.

Total number of vertices: 3749; Total number of edges: 57645

3. Traverse the graph data

The last section of this blog presents some examples of Gremlin traversals that take advantage of the speed of the Aerospike Database.

If you are new to Gremlin, this document, PRACTICAL GREMLIN: An Apache TinkerPop Tutorial, is a great place to start. I've added comments to make the queries easy to understand.

Query 1: Retrieve all columns for the airport with code 'AUS' (Austin)

The first query looks at all the data in the graph and filters vertices that have the 'AUS' airport code.

g.V()
  .has('code','AUS')
    .valueMap()
  .next()

g.V(): The vertex set of all the vertices in the graph.
.has('code','AUS'): Filtered by vertices with code = 'AUS'.
.valueMap(): Projects the result as a key value map of all the vertex properties.
.next(): Returns the final result set of the query.

display(pd.DataFrame(g.V().has('code','AUS').valueMap().next()))

Query 2: Display a pie chart of the ten top countries with the greatest number of airports

The next query demonstrates a count aggregation. It filters all the vertices labeled 'airport' and counts the number of airports by country.

g.V()
  .hasLabel('airport')
    .groupCount().by('country')
  .next()

.hasLabel('airport'): Create a vertex set of all the vertices with the vertex label equal to 'airport'.
.groupCount().by('country'): Group and count all occurrences of airports by country.

df = pd.DataFrame(g.V().hasLabel('airport').groupCount().by('country').next(), index=['Airports']).T
df.reset_index(inplace=True)
df.rename(columns = {'index':'Country'}, inplace = True)
srt_temp = df.sort_values(by = 'Airports', ascending=False)
not_top_ten = len(srt_temp) - 9
not_top_ten_sum = srt_temp.tail(not_top_ten).sum()
srt_top = srt_temp.head(9)
new_record = pd.DataFrame([{'Country':'Other', 'Airports':not_top_ten}])
df = pd.concat([srt_top, new_record], ignore_index=True)
df.sort_values(by = 'Airports', ascending=False, inplace=True)

plt.figure(figsize=(16,8))
ax1 = plt.subplot(121, aspect='equal')
df.plot(kind='pie', y = 'Airports', ax=ax1, autopct='%1.1f%%', 
 startangle=90, shadow=False, labels=df['Country'], legend = False, fontsize=10)

ax2 = plt.subplot(122)
plt.axis('off')
tbl = table(ax2, df, loc='center')
tbl.auto_set_font_size(True)
tbl.set_fontsize(10)
plt.show()

Query 3: Show possible routes from London to San Jose with one stop

This query limits the number of vertices returned using the limit() step.

g.V()
  .has('airport','code','LHR')
    .out().out()
      .has('code','SJC')
      .limit(15)
    .path()
    .by('code')
  .toList()

.has('airport','code','LHR'): Filters the vertices by those with the property 'airport' and 'code' value equal to 'LHR' (London Heathrow Airport).
.out().out(): Traverses two steps out from the previous filtered vertices. This means it retrieves vertices that are connected by edges going out two times from the 'LHR' vertex. These vertices represent flights departing from London Heathrow Airport, possibly making a stopover at another airport.
.has('code','SJC'): Filters the vertices retrieved in the previous step by those with the property 'code' equal to 'SJC' (San Jose International Airport).
.limit(15): Limits the result to only return a maximum of 15 vertices.
.path().by('code'): Retrieves the paths taken to reach the filtered vertices represented by the 'code' property of each vertex. This means it returns the flight routes from London Heathrow Airport to San Jose International Airport.

display(HTML("<style>.container { width:100% !important; }</style>"))
df = pd.DataFrame(g.V().has('airport','code','LHR').out().out().has('code','SJC').limit(15).path().by('code').toList())
df.rename(columns = {df.columns[0]:'Path'}, inplace = True)
legs = pd.DataFrame(columns=['Source_Airport', 'Dest_Airport'])
df = df.astype('str')
for index, row in df.iterrows():
    s = re.search(r"path\[(.*), (.*), (.*)\]", str(row))
    legs.loc[len(legs)] = [s.group(1),s.group(2)]
    legs.loc[len(legs)] = [s.group(2), s.group(3)]
df1 = pd.DataFrame(legs['Source_Airport'].unique(), columns = ['Airport'])
df2 = pd.DataFrame(legs['Dest_Airport'].unique(), columns = ['Airport'])
vertices = pd.concat([df1, df2])
vertices = pd.DataFrame(vertices['Airport'].unique(), columns = ['Airport'])
vertices_list =[(v, {"label": v}) for v in vertices.Airport.values.tolist()]
NX_graph = nx.DiGraph()
NX_graph.add_nodes_from(vertices_list)
NX_graph.add_edges_from(legs.values.tolist())

# Display the first 5 rows only in tabular form
display(legs.head(5))
nx.draw_networkx(NX_graph, node_size=1000, font_color='#FFF', node_color='#b20b00', pos=nx.spring_layout(NX_graph), with_labels=True)

Query 4: Return a map where the keys are the continent codes and the values are the number of airports in that continent

In this next query, use a child traversal that spawns anonymously from __. Usually, the function that gets this traversal will connect it to the previous traversal, as depicted in the following:

g.V()
  .hasLabel('continent')
    .group()
      .by('code')
      .by(__.out().count())
  .next()

.hasLabel('continent'): This filters the vertices to only include those with the label 'continent'.
.group(): This function groups the filtered vertices.
.by('code'): This specifies that the grouping key will be 'code' property of the vertices.
.by(__.out().count()): This specifies that the value for each group will be the count of outgoing edges from each vertex.

m = g.V().hasLabel('continent').group().by('code').by(__.out().count()).next()

fig,pie1 = plt.subplots()
pie1.pie(m.values() \
        ,labels=m.keys() \
        ,autopct='%1.1f%%'\
        ,shadow=True \
        ,startangle=90 \
        ,explode=(0,0,0.1,0,0,0,0))
pie1.axis('equal')  

plt.show()

Query 5: Find the number of international and domestic flights with one stop from SFO

The following query introduces four more steps. Namely, Project: “fold and unfold.”

g.V()
  .has("code", "SFO")
    .out().out()
      .dedup()
      .fold()
  .project("International Flights From SFO", "Domestic Flights From SFO")
    .by(__.unfold().count())
    .by(__.unfold()
      .has("country", "US")
      .count())
  .next()

.dedup(): Use the dedup step to remove duplicates from a result.
.fold(): Gathers all elements in the stream to that point and reduces them to a List.
.unfold(): Does the opposite, taking a List and unrolling it to its individual items and placing each back in the stream.
.project(): Similar to SQL project, that is, which properties we want to pick for display.

display_markdown(f''' ### {
g.V().has("code", "SFO").out().out().dedup().fold().project("International Flights From SFO", "Domestic Flights From SFO").by(__.unfold().count()).by(__.unfold().has("country", "US").count()).next()
}''', raw=True)

{'International Flights From SFO': 1904, 'Domestic Flights From SFO': 455}

Start exploring the world of graph

In this blog, I have shared a high-level view of the different layers of the Aerospike Graph Service and how those layers interoperate from the application layer to the database layer. I have also looked at the advantage of using the python_gremlin library to execute queries against the Aerospike database, which helps simplify the application development.

Aerospike Graph leverages open-source components, including the TinkerPop graph engine and the Gremlin graph query language. The current release initially addresses OLTP workloads, such as fraud detection and identity authentication, with OLAP functionality in the future. You can use it with many different programming languages; Python and Java are just two examples. Choose the language that works best for you.

I suggest you try it out. Get the most out of your data with an Aerospike Graph 60-day free trial.

In-memory database improvements with Database 7

Ronen Botzer — Wed, 15 Nov 2023 07:50:18 +0000

Welcome to Aerospike Database 7, and the initial server 7.0 release that sets the foundation for big developer API additions in server 7.1 and beyond. Server 7.0 also overhauls in-memory namespace in significant ways.

Aerospike releases have always been a continuum, with preceding minor releases providing prerequisite work upon which the next major release stands. For example, server 5.6 added a data structure for set indexes later used for secondary indexes (SI). Server 5.7 overhauled secondary index garbage collection and cut down SI memory consumption by 60%. Aerospike Database 6 was built on these essential improvements. Similarly, server 6.4 removed support for single-bin and data-in-index namespaces, freeing up primary index (PI) space needed for upcoming server 7.1 features.

Though major server releases have distinct themes, work on specific subsystems doesn’t end when our main focus shifts. Just as server versions 6.1 and 6.4 delivered significant throughput improvements to cross-datacenter replication (XDR) following the XDR rewrite theme of Aerospike Database 5, secondary index improvements will continue in Aerospike Database 7 releases.

Unified storage format revolutionizes in-memory namespaces

After previously using the same storage format for namespaces that persist data on SSD or Intel Optane™ Persistent Memory (PMem), an overhaul of in-memory namespaces in server 7.0 consolidates all three storage engines to the same efficient flat format. This results in multiple operational benefits.

Faster restarts for in-memory namespaces

In Aerospike Enterprise Edition (EE) and Standard Edition (SE), the new storage-engine memory places its data in shared memory (shmem) rather than volatile process memory. This means that in-memory namespaces can now fast restart (AKA warmstart) after clean shutdowns.

An in-memory namespace without persistence shares the fast restart capability mentioned above. More interestingly, cold restarts, during which Aerospike daemon (asd) rebuilds its indexes, run much faster, as record data is read from shared-memory, rather than a storage device.

An in-memory namespace with storage-backed persistence benefits from faster cold restarts when certain configuration parameters are adjusted (for example, partition-tree-sprigs). Only after a crash will Aerospike read from storage-backed persistence to repopulate record data into memory, along with rebuilding the indexes.

Adding to its existing capability of backing up index shmem segments to disk, the Aerospike Shared Memory Tool (ASMT) can now be used after Aerospike shuts down to back up in-memory namespace data ahead of restarting the host machine.

To summarize, in-memory namespaces without storage-backed persistence don’t lose their data on restarts of asd, and all in-memory namespaces benefit from faster restarts (both warmstart and faster coldstart).

Compression for in-memory namespaces

Now you can configure an in-memory namespace to use storage compression, such as ZStandard (zstd), LZ4, or Snappy. Customers already using storage compression in the persistence layer of an in-memory namespace (or data on SSD or PMem) will achieve the same compression ratio for their data, regardless of the storage engine, at the same CPU cost.

Stability and performance improvements for in-memory namespaces

As in-memory and on-device storage use the exact same storage format in server 7.0, an in-memory namespace is mirrored to its persistence layer. This means that its write-block defragmentation happens much faster in memory and no longer requires device reads. The same continuous write-block defrag mechanism eliminates heap fragmentation encountered in the former JEMalloc-based in-memory storage. Similarly, tomb raiding of durable-delete tombstones happens fully in memory and requires no device reads. This removes back-pressure generated by the persistence layer’s devices on in-memory namespace operations.

The unified format's single-pointer and contiguous record storage improve performance in two other ways. First, reading the entire record costs a single memory access, as opposed to the older scattered bins approach requiring multiple independent reads from RAM. Second, since write-blocks are mirrored to the persistence layer, write operations save the CPU previously consumed by separate serialization to a second storage format.

Capacity planning for in-memory namespaces

The first thing to note about in-memory namespaces with storage-backed persistence in server 7.0 is that the persistence layer is exactly the same as the in-memory storage; the two are mirrors of each other, which has a capacity planning implication. For previous versions, we had recommended that persistent storage be a multiple of memory-size (the max memory allocation for the namespace), typically 4x. Starting with server 7.0, persistent storage needs to be at a 1:1 ratio to the memory you wish to dedicate to your in-memory namespace.

The second thing to be aware of is that in-memory data storage is static, and gets pre-allocated in server 7.0 rather than progressively growing and bound by the (now obsolete) memory-size configuration parameter.

Capacity planning for indexes has not changed. In server 7.0, indexes continue to start small and grow in increments defined by the configuration parameters index-stage-size, sindex-stage-size. Set indexes grow in 4KiB increments after an initial pre-allocation. After upgrading a cluster node, the namespace indexes consume the same amount of memory as before, out of the system memory not pre-allocated for namespace data storage.

We no longer have diverging capacity planning formulas for data in-memory versus data on persistent storage. All storage-engine configurations use the same storage format and a single formula in the capacity planning guide. Treating capacity planning of in-memory data storage the same as you would data storage on an SSD is helpful. In both cases, you are aiming to size pre-allocated storage.

The memory consumed by your indexes (and room for them to grow), plus the pre-allocated namespace data storage, should fit within your host machine’s RAM and within the previously declared memory-size. Read the special upgrade instructions for Server 7.0 for more details.

Caveats

Support for single-bin namespaces was removed in server 6.4. Aerospike users without single-bin namespaces in their cluster may upgrade to server 7.0 through a regular rolling upgrade. Otherwise, please consult the special upgrade instructions for Server 6.4.

Due to the unified storage format, a write-block-size limit of 8MiB applies to in-memory namespaces (with or without persistence). Aerospike users who depend on the former 128MiB record size limit of in-memory without persistence will need to break up their records. Customers may choose to delay upgrading till server 7.1, which will enable an easier transition.

Configuration and monitoring

The newly released Aerospike Observability Stack 3.0 and Tools 10.0 support metrics and configuration for both Aerospike 6 and Aerospike 7 and are designed to ease your transition to server 7.0. If you are not familiar with Aerospike Observability and Management (O&M), we have a short video, blog, and webinar available on our site.

Configuration parameter and metric changes in detail

Namespace configuration, regardless of storage engine choice, is simpler in server 7.0.

Stop-writes and eviction thresholds are controlled by the storage-engine configuration parameters stop-writes-used-pct, stop-writes-avail-pct, and evict-used-pct, which are relative to the namespace data storage size. There are also a pair of thresholds relative to the system memory - stop-writes-sys-memory-pct and evict-sys-memory-pct.

Data storage metrics have been simplified to data_used_bytes, data_total_bytes, data_used_pct, and data_avail_pct for all storage engine types.

Index metrics are also simpler. Set indexes have set_index_used_bytes. Primary and secondary indexes have index_used_bytes and sindex_used_bytes, whether they’re stored in shared memory, PMem, or an SSD. If they’re in persistent storage, they also have index_mounts_used_pct and sindex_mounts_used_pct relative to the mounts_budget.

Multi-tenancy

Many Aerospike customers deploy their database clusters as a multi-tenant service, with distinct users separated by sets within a namespace. Multi-tenancy leans on Aerospike enterprise features, such as scoped role-based access control (RBAC), rate quotas, and set quotas.

Server 7.0 makes multi-tenant deployment easier with several new features.The limit of 64K unique bin names per-namespace was removed. Operators no longer need to advise developers to restrict how many bin names their applications write into the namespace. As a result, the bins info command and the available_bin_names namespace statistic, were removed.

The limit on unique set names per namespaces was raised from 1023 to 4095, allowing for set-level segregation of more tenants on the same Aerospike cluster.

Finally, an operator can now assign a unique set-level default-ttl as an override of the namespace default-ttl.

New developer API features

Server 7.0 adds the capability to index and query bytes data (BLOBs).

Application developers may now choose to persist key-ordered Map indexes, trading off extra storage for improved performance. The new MapPolicy option can be applied when creating a new Map bin or with the Map set_type operation.

MapPolicy(MapOrder order, int flags, boolean persistIndex)

Persisting Map indexes should only be used by an application once all the nodes in the Aerospike cluster have been upgraded to version >= 7.0.

Dropping support for old OS versions

Server 6.4 added support for Amazon Linux 2023 and Debian 12. As I previously warned, server 7.0 removes support and el7 builds for Red Hat Enterprise Linux 7 and its variants, including CentOS 7, Amazon Linux 2, and Oracle Linux 7. Similarly, server 7.0 will not be available on Debian 10.

Aerospike engineering takes performance seriously, and Aerospike users choose to build mission-critical applications on our database because it has the best cost-performance in its field. Performance is hindered by running new Aerospike server versions on ancient lower-performing OS kernels, such as the 3.10 Linux kernel packaged with RHEL 7. In a recent announcement, the Linux kernel team declared they will no longer offer six years of LTS support. This announcement validated our perspective on the stability and performance cost of running Aerospike on old kernels.

Both Debian 10 and RHEL 7 reach their end of life (EOL) in June 2024 ( 5 and 10 years from their initial release, respectively ). Each new major and minor Aerospike server version gets two years of bug fixes and security vulnerability support. Going forward, new server versions will not be offered on OS distro versions scheduled to expire during this support period. Subsequent patch releases (hotfixes) will continue to be built and tested on the same OS distro versions as when they were first released.

Try Aerospike 7.0

Persistence, fast restart capability, and built-in compression all make the new Aerospike in-memory namespaces appealing for in-memory database use cases.

Consult the platform compatibility page and the minimum usable client versions table. For more details, read the Server 7.0 release notes. You can download Aerospike EE and run in a single-node evaluation; you can get started with a 60-day multi-node trial at our Try Now page.

Forem: Aerospike

Introducing Aerospike Graph Database 3.0: Faster, simpler, and built for the terabyte scale era

Easier to develop

Multi-property support

Native Datetime support

Faster to load

Cheaper to run

Get started

How Criteo scaled 290M QPS and cut its server footprint by 78%

Maintaining real-time access at AdTech scale

The legacy stack: Couchbase + Memcached + custom C logic

Kubernetes-native deployment: From custom scripts to automated ops

Multi-bin optimization: Collapsing data models for performance

Infrastructure impact: 78% server reduction and lower carbon footprint

Lessons learned: Performance tuning, not over-engineering

Takeaway: Scaling real-time systems without scaling costs

Aerospike Rust client preview: Introducing a native integration for high-performance systems

Why Rust: Strategic alignment with developer needs

Rust client capabilities

Release timeline

Get started

Pi day exercise: Calculating pi with expressions!

What is Pi?

Calculating the value of Pi

Get the most out of your data with Aerospike Expressions

How to retrieve the key in a key-value store

Keys in Aerospike

Storing keys in records

Key mismatch explained

User key stored with the record

What are Aerospike Expressions?

Key retrieval code example

About Value.NULL

Case 1: key type known

Case 2: key type unknown

Unlock the power of expressions in Aerospike for enhanced data operations

Using Aerospike policies (correctly!)

What are policies?

Networking control policies

Timeout delay

Default policies

Creating policies

Summary

Unlock Aerospike Cloud: Exploring the power of the Aerospike LINQPad driver

Connection dialog

Driver overview

Using the Aerospike null set

Samples

Review

How to use the Kubernetes Operator to establish connectivity between clusters

Setup

Destination Cluster

XDR-Proxy

Configuration

Deployment

Source cluster

Simple message test

Interim summary

Scaling the XDR-Proxies

Add data to the source cluster

Achieve dependable resiliency with Aerospike

Using auto-values in Aerospike LINQPad driver

Example set

Simple cast examples

Handling unsupported data types

Conversion

Collection data types

Primary key

LINQPad driver samples

Using AWS IAM for client authentication

AWS MKS Kafka

Kafka client machine

Kafka topics

Creating topics

Listing topics

Producer and consumer

Database source

Create the Aerospike Database

Install the Aerospike Database server

Aerospike Database configuration file for use with systemd

About `Value.NULL`

Aerospike Database configuration file for use with `systemd`