Coherence SIG - Filtering 100M objects in cache

Today I was speaking on Coherence SIG event in London.

My topic was "Filtering 100M objects. What can go wrong?". It was a story of solving particular problem and obstacles we have encountered. One noticeable thing about this project - out team was using Performance Test Driven Development approach.

We have started with simplest solution, then were focusing on problem identified by testing.

Slide deck from presentation is below.

Data Grid Pattern - Canary key set

Most of modern in-memory-data-grid products have grown from distributed caching. For traditional cache, loss of data is not a big deal, missing data could always be recovered from backend storage via read through. Main focus of distributed cache is data coherence between nodes (preventing stale reads, etc).

Advanced patterns like all-in-memory and proactive caching may provide considerable benefits over traditional read through cache. But simple fall back to read-through as a strategy for data recovery may not be an option for these advanced cache usages. Read-through have two prerequisite to be implemented:

• data should be accessed by primary key,
• for given primary key, you should known its master data source.

Both these prerequisite may be broken in advanced solution. Having all-data in memory will allow you to execute queries in cache layer instead of backend database. Products like Oracle Coherence, GemStone GemFire and GigaSpaces have advanced queering capabilities (including non-primary indexes support). Offloading queries from database is a huge win, but the price is that you cannot rely on read-through any more. If some data is missing in cache, queries will produce incomplete results without warning.

Second read-through prerequisite also may be sacrificed, i.e. by using multiple backends (cache acting as aggregator for data set scattered across multiple databases). You can more details in my previous article.

Data loss imminent

Please mention, that loss of data in modern in-memory-data-grid is an exceptional event. Data is usually protected by multiple replicas and grid can tolerate server failure. But it still possible and you cannot ignore this aspect as you cannot ignore e.g. backing up your database.

Through all reliability provided by data grid technology data may be lost and it means they will be lost eventually. Next question, what is your desired strategy to cope with incomplete data set?

It depends on type of application.

•  For some applications, it is ok to have incomplete results from application during recovery window.
• For some application, incomplete response is worst than no response. Application should guaranty that every response is complete and if it cannot provide complete response (e.g. part of data set is missing in cache) it should raise an error.

First strategy is rather simple, you should monitor your grid and automatically trigger recovery procedure if disastrous event is detected.

Second type of strategy is more tricky in implementation. Just monitoring is not an option.  There would be a gap between data loss event and reaction of monitoring system (which e.g. can switch service to offline mode for duration recovery process). Some application are totally intolerant to inconsistent data. This data, for example, could be used in complex batch of financial risk calculations (running for few hours in large HPC grid) and single inconsistent piece of input data could invalidate whole batch of work.

We need a solution better than monitoring for such kind of applications.

Canary keys to detect missing data

We must guaranty that result of each query is consistent (i.e. all data that has to be processed has been processed, often this means - whole dataset). Here we have a paradox at hands: we must check presence of  certain key/value pairs (data grid is a key/value storage) in cache, but we cannot know keys of these pairs or even their total number.

Solution to this paradox lies in base approach of data distribution used by data grids. Technique described in this article has been used by me with Oracle Coherence grid, but idea can also be applied to other products using similar type of DHT. Oracle Coherence is using partitioned distributed hash table (DHT). In practice this means that key/value pairs are not distributed individually, but whole partition is assigned to particular node. It also means, that you cannot lose individual key/value pair but only whole partition at once (if you lose all replicas of that partition at the same time). Number of partitions is fixed for live time of grid (changing that number requires rebuild of DHT).

How this could be helpful to ensure data completeness?

We may not be caring about presence of individual key/value pairs, instead we may check presence of all partitions (and we know their exact IDs and total number). But how we can check presence of partition (technically partition cannot be missing, it will be just empty)? Also we should join data completeness check with queering of data in the same operation otherwise we will always have a gap of uncertainty.

Canary keys is a trick to solve this problem. Canary keys are synthetic keys, you put just on key to each partition. Every partition should have a canary key. So if your grid is configured to have N partitions, it should contain exactly N canary keys. If number of canary keys is less than N, that means portion of data has been lost (poor canary has perished) and is not recovered yet. Of cause your data loading/recovery procedure should put canary keys back in cache once data is restored.

It is also possible (though quite awkward to be honest) to integrate canary keys check in single request with data queering. In each query you should select canary keys along with actual data you need to retrieve. Once query is executed you should check presence of all canaries and then strip them from result set. If all of them are there, you can be sure that your result set in complete and no SLA would be broken. If some canaries are missing, you should trigger data recovery and/or raise an error indicating that request cannot be completed until recovery is done.

Conclusion

Technique described in this article is very advanced. Most applications do not require such rigorous consistency checks for every operations. But few do, and for them this approach may be useful (still implementation may have its quirks).

On high level canary keys technique demonstrate how understanding of  core principles of DHT can help solving challenging task. Understanding of low level grid operations, their guaranties and limitation is a cornerstone in engineering of complex data processing solution using distributed ACID-less data grid as a storage.

See also

Open source implementation of canary keys framework - http://code.google.com/p/gridkit/wiki/DataLossListener.

Data Grid Pattern - Time series index for managing versioned data

Many critical application are using append only approach for dealing with transactional data. In other words, they never update data records, but instead insert new records with greater timestamp (or sequence number, or any other kind of version, they are using to find latest record). Common challenge for such data model, is how to retrieve an appropriate version of record (e.g. latest version, or version at certain moment in time). A simple query for latest version for key 'A' would translate into query as complex as

select * from versions where series='A' and version = (select max(version) from versions where key='A')

This query is already too complex for data grid (and even RDBMS would not be too happy).

Accidently Ben Stopford has recently published a great article about this problem, outlining a lot of important aspects. I do not want to repeat him here, I suggest you read his article now, then continue with mine, which complements his two approaches this third one using custom index in Coherence.

Using custom index for accessing versioned data

In this  approach, each version is stored as a separate entry in cache.

Entry key is composite key, including logical key (series key) and some additional field to make version key unique (e.g. transaction ID, sequence number etc). Value contains actual business data (payload) and timestamp, we are using in our queries (technically timestamp could be part of composite key).

Series key should be an affinity key also - all versions related to same series should be physically on one Coherence node (or affinity key can be a part of series key, this will also satisfy this requirement).

Normally, if you want to find certain version by series key and timestamp you have to do aggregation of all versions for this series. In two approaches mentioned by Ben Stopford, latest version is separated from all other versions (using separate cache - approach 1, using marker - approach 2). It solves problem of finding latest version, but doesn't help if we need to find version for certain moment at time.

Time series index structure

Normal Coherence indexes cannot help us much, due to complexity of query, but it is possible to create custom index, tailored specifically for this task.

Time series index is similar to traditional inverted index, but instead of storing set of entry references, it is storing a nested index, indexing  only versions belonging to certain series by timestamp. Using this index structure you could find latest version or version for certain moment in time, without any aggregation, just by index lookup.

This index goes beyond standard index Coherence API, so it requires a complementary implementation of custom filter.

PROs and CONs

Below and PROs and CONs of this approach, compared to approaches from Ben's article.

PRO

• Inserting new version doesn't require modifications of any other versions. In particular, you do not need to use hack, directly accessing to backing map, and you do not create extra replication traffic.
• Time series index works efficiently for any point in time, not only latest versions.
• It can be used with any kind of caches (even with continuous queries).

CON

• Through custom index usage is straightforward, troubleshooting could be very tricky unless you understand index mechanics very well.

Source code

Time series index implementation is available at GridKit project.

Data Grid Pattern - Proactive caching

Classic and most widely used approach for caching is read through pattern. Look up in cache, then try to load from primary data source if entry is missing in cache - that is how it works. This pattern is easy to implement but it has few unpleasant limitations:

• caching may reduce average response time, but maximum response time is still bound to back end data source response time,

• cache may have stale data, expiry policy may relive this problem to some extent, but aggressive expiry is drastically reducing performance gain from caching.

All in memory pattern

Caching concept is close relative to memory hierarchy principle. Memory hierarchy is one of cornerstones of Von Neumann architecture relies on fact that we have different kinds (in terms of capacity and performance) of memory in system.  With modern hardware dynamic memory capacity is often large enough to keep whole dataset. All-in-memory is term used to describe caching architecture there you have 100% of your data in cache at all times. Having all data in cache allow you to guaranty that no request will have to hit slow backend data source and thus provide more aggressive SLA for max response time. It also may be required for workloads with highly random access to data, there traditional assumptions like 80/20 are not working.

While all-in-memory approach is definitely win in terms of performance, its implementation has few serious challenges:

•  cache should have enough capacity to hold 100% of our data,
• cache should be fault tolerant (losing a portion of data in cache will render it defunct until missing data would be reloaded),
• preloading procedure is often non-trivial due to scale of data set,
• cache should be kept in sync with backend data source.

Capacity and fault tolerance are provided by modern distributed caches out of box, but preloading procedures and cache update strategy are very application specific and fairly challenging to implement.

Below are few practical approaches for keeping cache in synch with primary data source.

Refresh ahead

This is approach similar to expiry policy, but instead of invalidating data, cache proactively refreshing them from master source. Refresh ahead pattern is quite simple, but not very practical though. If cached data set is large (and we are talking about all-in-memory pattern) automatic refreshing is like to overwhelm backend with requests. And even if data set is reasonably small we still have to use fairly long expiry time to make it practical.

So if you looking for all-in-memory cache, refresh ahead are unlikely to help you.

Proactive caching

In contrast with traditional (I would say reactive) caching, with proactive caching pattern you insert/update value in cache as soon as it is updated in backend data source, not at the moment data was requested from cache.

Proactive caching is not necessary should be used with all-in-memory pattern, but combination of these two is very powerful.

Polling updates from DB

An evolutionary step from refresh ahead to proactive caching, would be polling changes for database. Some daemon component should periodically (and frequently) poll database and fetch changes since last cache update. Sounds simple but you have to come out with a way how to "fetch changes since last cache update".  Usually some kind of timestamp is used - each record in table to be cached has a kind of last modified field. Another few challenges:

·         what if cache is representing a query result, not just a single table,

·         'last modified' field should be indexed, otherwise frequent polls will bring database to its knees,

·         polling daemon should be fault tolerant,

·         cache timestamp should be stored somewhere on cache side.

So, implementing this approach will require some amount of work (on both sides, database and cache), but in the end you will have very robust solution.

Only sever limitation of this approach is lag between changes in database and cache, which is no less than poll period.

Long poll

If database has some means for wait/notification in its query language, you can use long poll pattern. 

Using long poll will reduce load on your database server and probably reduce lag between cache and database. Disadvantages of this approach: more code on database side and need to use dedicated thread(s) for polling (because thread will be blocked waiting notification of database side for most time).

In Oracle database long poll could be implemented using DBMS_ALERT package. Though use of DBMS_ALERT may cause serialization of update transaction and harms database performance.

Database notifications

Some databases can push data change notifications directly to clients, without need for polling. E.g. Oracle database has DCN (data change notification) mechanism. Using DCN you can register callbacks which would be notified that certain data have been changed in database. Notification mechanism has few advantages over polling

• less load of database while no data is actually changing,
• smaller lag between changing data in database are reaction in cache.

Notifications approach have disadvantages also

• API usually more complicated, you have to learn more quirks to make it work,
• connection hang problems - application is listening to event on connection which is defunct for some reason,
• notifications may be lost in transition for some reason.

One particular problem with using Oracle DCN in java, was leaking of subscriptions. DCN subscription is remaining active on database side after termination java process (unless it was deregistered explicitly) and eventually you are going to hit limit for active subscriptions. Of cause you should deregister subscription before terminating of client process, but you cannot always guaranty graceful shutdown in practice.

Hooking into database replication

All mature databases have replication feature and thus replication wire protocol. Sometimes, they also provide API to hook into replication channel and programmatically receive all updates (essentially change notifications).

Replication is implemented differently in various databases (or even with different replication solution for same database). But general idea is to make cache act as replication slave for its source database.

Compared to polling or data change notifications, using of replication usually requires more effort from DBA side (they should setup replication slave of master database). It may also cost you some in licensing fees dependent on your replication solution.

MySQL slave protocol does not require any setup on master, so replication links can be created ad hoc. But MySQL has another catch, you should setup row based replication on master database unless you want to parse and execute SQL statements in your cache.

Single cache in front of multiple sources

In large scale system, primary data source may be distrusted itself  (e.g. using sharded database). Having single read through cache may be a problem in this case (doing read through, you have to know which shard to consult about data missing in cache), but with proactive caching such setup it much straightforward. While in read through caching, cache responsibilities of serving requests and acquiring data are coupled. With proactive caching, responsibilities of storing data/serving read requests and feeding cache with data updates may be separated. This way, you can have single cache instance and multiple other components pushing data into it (e.g. each sharing could push its data in single cache).

In this role cache can be though as a kind "materialized view" based up on data in primary (potentially distributed) data source.

Few more links

·         Coherence SIG: Using Oracle Coherence to Enable Database Partitioning and DC Level Fault Tolerance

Using Database Change Notification (DCN) with a Coherence Cache

Data Grid Pattern - Snowflake data schema

Evil of distributed joins

Traditionally data grids based solutions are using denormalized data models. Denormalized model allows reducing number of data lookups in storage and achieve low response time. Traditional normalized data models are no good for distributed key/value storages. Main argument against them is requirement to do multiple joins during typical data access operation. If your data are partitioned, joins are becoming prohibitively expensive. You have to join each partition from left join side with each partition on right side thus cost of join grows in quadratic proportion of your number of partitions. 

Even if you can narrow record set before actual join, your data is still may be located on different servers. Each join between partitioned tables will force your data to be moved across network, adding network latency to response time. More joins – more latency accumulated and you will find you out of your allowed response time very soon.
I other words, generally, in grid, you cannot use neither joins, nor normalized data model (though there may always be an exception).

Denormalization is evil

But data denormalization approach is far from perfect. It has its downsides and serious ones. Data redundancy is a side effect of denormalization. Redundancy increase memory consumption and cost of updates. Loosing consistency due to redundancy is another foe. It is a constant battle.

Data warehousing experience

Data warehousing industry has similar challenges to solve. They have huge amounts of data and also have to deal with distributed data storages. And they have an answer to this challenge for some time!

Analytic databases are usually using "snowflake" data model. Snowflake schema has single fact table (in center of schema) and several dimension tables. Size of fact table is huge; sizes of dimension tables are relatively small.

Below are few examples to illustrate snowflake schema.

From financial industry

and from retail

Snowflake data model allows implementing joins across distributed data storage in smart way. While we still want to have fact data to be partitioned across cluster, we can have a full copy of dimension tables on each node executing queries. In other words we may use replication strategy for small dimension table while keeping large fact data partitioned.
Now we can partially execute query using dimensions which are available in local memory of process and then issue single query to fact table. As long as you do not need to join “facts” table with itself of another facts table query execution requires just one network round trip (and if you really need joins between facts wait for next article “Map/Reduce in data grid”).

How can we you utilize snowflake schema with grid middleware?

First we have to decide how we are going to store dimensions tables in memory. We can use in-memory data grid itself or we can use some in-memory database like Hypersonic or H2 database. Grid will support replication of dimension data out of box and natural support for java object, but in memory relational DB will offer better support for queries (e.g. Oracle Coherence does not support any kind of joins even in replicated storage).
Next question is cluster topology, we may replicate dimension data over all nodes or just few of them (query nodes). These query nodes can be either grid peer or remote grid client.
Functional separation between query nodes and data nodes (one storing partitions of facts table) is probably most practical because they have different memory/CPU usage. This way you can tune JVM memory options and scale each tier independently.

Unfortunately this is just a pattern but not built in feature any data grid product I'm aware of. You still have to complete a lot of non trivial engineering to make it work, but this work worth an effort. This pattern allows to use IMDG in cases they denormalization is unable to solve problem, opening new horizons for using of technology.

Data Grid Pattern - Network shared memory

Using shared memory for inter-process communications is a very popular pattern in UNIX world. Many popular RDBMS (both open source and commercial) are implemented as sets of OS processes communicating via shared memory. Designing a system as a set of functionally specialized processes cooperating with each other helps to keep system more transparent and maintainable; while using shared memory as IPC keeps communication overhead is very low. Shared memory is very efficient compared to other forms of IPC. Several processes can exchange or share information with each other via shared memory without involvement of OS (e.i. no syscalls, no context switching overheads).

Unfortunately shared memory is useful only if all communicating processes are hosted on same server (e.i. connected to same memory circuits). While we cannot use real shared memory when building a distributed system, idea of building system as a set of cooperating specialized processes still remains attractive.

Key point of shared memory is what all process can access and modify same data; all processes have consistent view and can do atomic operations on data (e.g. compare-and-set). Data grid technologies allow us to achieve same features but in network environment. Modern data grid products (e.g. Oracle Coherence, GemStone GemFire, etc) provide us key features for shared-memory-like communication:

• share data with strong consistency guaranties across processes in cluster;
• atomic operations over single key in shared storage.

Still, data grids remain a very different technology. They have with different tradeoffs compared to shared memory. It is impossible to take a PostgresSQL and make it clustered using Oracle Coherence. But we can reuse and extend architectural approach and develop a “network shared memory” pattern as a way to build distributed system.

“Network shared memory” pattern

First you need separate data and processes in your mind. Next you should identify operations with data which are happening in your system. You may have operations such as serving request using data, updating, importing data, exporting data, transforming data, etc. Ideally you should have separate process dedicated for each operation (though it is not always possible due to various reasons, so be holistic). Once you finished with analyzing your data and designing processes, you can start to work on physical model to store your data in grid. Remember, data for grid should be always modeled with access pattern in mind. Just coping data model from RDBMS may produce disastrous results (data modeling for grid is another large topic).

 Why having multiple processes is better?

Why having multiple processes is better compared e.g. with multiple threads within a single process?

Below are few answers:

• You can distributed your processes between servers (e.g. for optimizing resource utilization)
• You can start/stop processes independently
• You can upgrade processes independently
• You have better failure isolation (e.g. memory leak in one process will not bring down whole system)
• You can use less heap per JVM, and have individual memory options for different processes

Sure there are some drawbacks also. A few of them:

• You will have more JVMs running and thus more overhead of JVM itself.
• All processes in the end have to communicate over network and it is not as fast as using memory shared between threads.

 Advanced usages of “Network shared memory”

Data grids have high availability built in. We can leverage this feature to achieve high availability for our processes. First pattern is a “hot standby”. We may have several instances of same process running (e.g. on different boxes) but only one of them active. In case of active process going down, grid will detect process failure and we can promote one of standbys to be new active. Sounds simple, but this simple approach requires “death detection”, “peer discovery” and “distributed consensus”. Believe me, all of these is not an easy task to implement. Fortunately, data grid already have all of this implemented, you just need to use its API. Data grid also can be used to store internal state of process, this way you can implement failover even for stateful processes.

A next evolutionary step of this approach is a load balancing between processes. Data grid can help coordinating processes by sharing routing table for request, state for stateful operations and/or using distributed locks for controlling access to resources.

Data grid is more than just distributed data store

While data grid technology is primarily designed for working with larger data sets, advanced features of modern data grid products may bring benefits to your system even if all your data fit a memory of single server. Their high availability and distributed coordination features may be invaluable for designing modular distributed solutions.

Data Grid Pattern - Data flow mediator

I want to start series of articles about data grid oriented architectural patterns. First pattern I want to present is a “Data flow mediator”. Let me start with example.

Imagine you have a large ecommerce web application and want to do some real time analysis over user actions. You have a stream of simple event, let’s just say clicks. And you need to do real time aggregation by various dimensions like by user, by product, etc. At large scale this task is quite challenging: number of writes is enormous, different dimensions made shading challenging and business want this analytics as close to real time as possible (say few seconds delay). With or without data grid such will remain challenging, but data grid technology have a strong advantages for such task.

Let me now introduce “data flow mediator” pattern.

In this pattern, data grid is used as buffer between systems which produces events (clicks), and systems which consumes information (real time analysis modules).

From producer point of view:

• Grid provides high and scalable throughput,
• Grid provides reasonable balance between durability/performance/cost. In grid we can store data in memory only, protected by multiple redundant copies.

From consumers’ (RT analysis modules) point of view:

• Advanced data grids (e.g. Coherence, GemFire, etc) provide required queering/aggregation tool (implementing efficient queries by multiple dimensions in grid still an art, but it is doable),
• High and scalable read throughput. Different analysis modules may share same “mediator”

From architect point of view:

• Mediator decouples data producer from data consumers, thus localizing impact of changes for each component,
• Data grid is self managing. Imagine managing DB with 50 shards + HA replication + dynamic adding/removing servers to cluster and you will treasure this feature of data grid.

I have demonstrated this pattern with ecommerce examples, but there are similar use cases in finance and telecom. Key prerequisites for this pattern are:

• Large number of small updates,
• Data loss is not fatal (either we can tolerate it or restore data from somewhere else),
• Large number of read queries,
• Queries are reasonable simple but more complicated than just get by primary key,
• Low response time requirements for both read and write,
• Scale is above than single RDMB can handle.

I hope this article was helpful for you to better understand data grid technology and its use cases.