GridGain - In-Memory Big Data

Java Serialization - Good, Fast, and Faster

2012-12-07T12:26:00.002-08:00

Probably anyone who has ever worked with serialization of objects, be that in Java or any other language, knows that it should be avoided whenever possible. Just like the first rule of distribution is "Do not distribute!", the first rule of serialization should be "Do not serialize!". However, in many cases, especially in distributed environments, serialization cannot be avoided and therefore must be significantly optimized to achieve any kind of reasonable throughput.

At GridGain, given the distributed nature of our product, we have always been working on optimizing of our serialization routines, but starting with version 4.3.0 we have achieved the fastest results so far. Our GridOptimizedMarshaller in our tests achieved up to 20x performance optimization on standard Java serialization with java.io.Serializable. If you switch to java.io.Externalizable, then GridGain marshaller is up to 10x faster. We have even compared our marshaller to Kryo serialization, and turns out that our marshaller is up to 5x faster than Kryo. On top of that, the footprint of GridGain serialized objects is significantly smaller than Java.

The coolest thing here is that we do not require any custom interfaces or API s - GridGain optimized serialization works directly with standard Java POJOs, regardless if they implement java.io.Serializable interface or not. If your POJOs implement java.io.Externalizable, then our marshaling works even faster.

How do we do it? The main culprit of Java serialization is java.io.ObjectOutputStream which is extremely expensive to initialize and performs poorly. The first thing we did is replaced it with our own implementation, based on direct memory copying by invoking native C and Java so-called "unsafe" routines. We also serialize fields in predefined order by doing lots of object introspection which allows us to pass only values and not their type names or other metadata.

Here are the results from the test on my MacBookPro 2.7 GHz Intel i7:

>>> Java serialization via Externalizable (average): 22,551 ms
>>> Kryo serialization (average): 17,300 ms
>>> GridGain serialization (average): 2,937 ms

Here is the test itself. This test is included with our product, so feel free to download it and try for yourself.

public class SerializationBenchmark {
    /** Number of runs. */
    private static final int RUN_CNT = 3;

    /** Number of iterations. */
    private static final int ITER_CNT = 200000;

    public static void main(String[] args) throws Exception {
        // Create sample object.
        SampleObject obj = createObject();

        // Run Java serialization test.
        javaSerialization(obj);

        // Run Kryo serialization test.
        kryoSerialization(obj);

        // Run GridGain serialization test.
        gridGainSerialization(obj);
    }

    private static long javaSerialization(SampleObject obj) throws Exception {
        long avgDur = 0;

        for (int i = 0; i < RUN_CNT; i++) {
            SampleObject newObj = null;

            long start = System.currentTimeMillis();

            for (int j = 0; j < ITER_CNT; j++) {
                ByteArrayOutputStream out = new ByteArrayOutputStream();

                ObjectOutputStream objOut = null;

                try {
                    objOut = new ObjectOutputStream(out);

                    objOut.writeObject(obj);
                }
                finally {
                    U.close(objOut, null);
                }

                ObjectInputStream objIn = null;

                try {
                    objIn = new ObjectInputStream(
                        new ByteArrayInputStream(out.toByteArray()));

                    newObj = (SampleObject)objIn.readObject();
                }
                finally {
                    U.close(objIn, null);
                }
            }

            long dur = System.currentTimeMillis() - start;

            avgDur += dur;
        }

        avgDur /= RUN_CNT;

        System.out.format("\n>>> Java serialization via Externalizable (average): %,d ms\n\n", avgDur);

        return avgDur;
    }

    private static long kryoSerialization(SampleObject obj) throws Exception {
        Kryo marsh = new Kryo();

        long avgDur = 0;

        for (int i = 0; i < RUN_CNT; i++) {
            SampleObject newObj = null;

            long start = System.currentTimeMillis();

            for (int j = 0; j < ITER_CNT; j++) {
                ByteArrayOutputStream out = new ByteArrayOutputStream();

                Output kryoOut = null;

                try {
                    kryoOut = new Output(out);

                    marsh.writeObject(kryoOut, obj);
                }
                finally {
                    U.close(kryoOut, null);
                }

                Input kryoIn = null;

                try {
                    kryoIn = new Input(new ByteArrayInputStream(out.toByteArray()));

                    newObj = marsh.readObject(kryoIn, SampleObject.class);
                }
                finally {
                    U.close(kryoIn, null);
                }
            }

            long dur = System.currentTimeMillis() - start;

            avgDur += dur;
        }

        avgDur /= RUN_CNT;

        System.out.format("\n>>> Kryo serialization (average): %,d ms\n\n", avgDur);

        return avgDur;
    }

    private static long gridGainSerialization(SampleObject obj) throws Exception {
        GridMarshaller marsh = new GridOptimizedMarshaller(false, 
            Arrays.asList(SampleObject.class.getName()), null);

        long avgDur = 0;

        for (int i = 0; i < RUN_CNT; i++) {
            SampleObject newObj = null;

            long start = System.currentTimeMillis();

            for (int j = 0; j < ITER_CNT; j++)
                newObj = marsh.unmarshal(marsh.marshal(obj), null);

            long dur = System.currentTimeMillis() - start;

            avgDur += dur;
        }

        avgDur /= RUN_CNT;

        System.out.format("\n>>> GridGain serialization (average): %,d ms\n\n", avgDur);

        return avgDur;
    }

    private static SampleObject createObject() {
        long[] longArr = new long[3000];

        for (int i = 0; i < longArr.length; i++)
            longArr[i] = i;

        double[] dblArr = new double[3000];

        for (int i = 0; i < dblArr.length; i++)
            dblArr[i] = 0.1 * i;

        return new SampleObject(123, 123.456f, (short)321, longArr, dblArr);
    }

    private static class SampleObject 
        implements Externalizable, KryoSerializable {
        private int intVal;
        private float floatVal;
        private Short shortVal;
        private long[] longArr;
        private double[] dblArr;
        private SampleObject selfRef;

        public SampleObject() {}

        SampleObject(int intVal, float floatVal, Short shortVal, 
            long[] longArr, double[] dblArr) {
            this.intVal = intVal;
            this.floatVal = floatVal;
            this.shortVal = shortVal;
            this.longArr = longArr;
            this.dblArr = dblArr;

            selfRef = this;
        }

        // Required by Java Externalizable.
        @Override public void writeExternal(ObjectOutput out) 
            throws IOException {
            out.writeInt(intVal);
            out.writeFloat(floatVal);
            out.writeShort(shortVal);
            out.writeObject(longArr);
            out.writeObject(dblArr);
            out.writeObject(selfRef);
        }

        // Required by Java Externalizable.
        @Override public void readExternal(ObjectInput in) 
         throws IOException, ClassNotFoundException {
            intVal = in.readInt();
            floatVal = in.readFloat();
            shortVal = in.readShort();
            longArr = (long[])in.readObject();
            dblArr = (double[])in.readObject();
            selfRef = (SampleObject)in.readObject();
        }

        // Required by Kryo serialization.
        @Override public void write(Kryo kryo, Output out) {
            kryo.writeObject(out, intVal);
            kryo.writeObject(out, floatVal);
            kryo.writeObject(out, shortVal);
            kryo.writeObject(out, longArr);
            kryo.writeObject(out, dblArr);
            kryo.writeObject(out, selfRef);
        }

        // Required by Kryo serialization.
        @Override public void read(Kryo kryo, Input in) {
            intVal = kryo.readObject(in, Integer.class);
            floatVal = kryo.readObject(in, Float.class);
            shortVal = kryo.readObject(in, Short.class);
            longArr = kryo.readObject(in, long[].class);
            dblArr = kryo.readObject(in, double[].class);
            selfRef = kryo.readObject(in, SampleObject.class);
        }
    }
}

GridGain and Hadoop: Differences and Synergies

2012-11-20T12:03:00.000-08:00

Overview

GridGain is Java-based middleware for in-memory processing of big data in a distributed environment. It is based on high performance in-memory data platform that integrates fast In-Memory MapReduce implementation with In-Memory Data Grid technology delivering easy to use and easy to scale software. Using GridGain you can process terabytes of data, on 1000s of nodes in under a second.

GridGain typically resides between business, analytics, transactional or BI applications and long term data storage such as RDBMS, ERP or Hadoop HDFS, and provides in-memory data platform for high performance, low latency data storage and processing.

Both, GridGain and Hadoop, are designed for parallel processing of distributed data. However, both products serve very different goals and in most cases are very complementary to each other. Hadoop is mostly geared towards batch-oriented offline processing of historical and analytics payloads where latencies and transactions don't really matter, while GridGain is meant for real-time in-memory processing of both transactional and non-transactional live data with very low latencies. To better understand where each product really fits, let us compare some main concepts of each product.

GridGain In-Memory Compute Grid vs Hadoop MapReduce

MapReduce is a programming model developed by Google for processing large data sets of data stored on disks. Hadoop MapReduce is an implementation of such model. The model is based on the fact that data in a single file can be distributed across multiple nodes and hence the processing of those files has to be co-located on the same nodes to avoid moving data around. The processing is based on scanning files record by record in parallel on multiple nodes and then reducing the results in parallel on multiple nodes as well. Because of that, standard disk-based MapReduce is good for problem sets which require analyzing every single record in a file and does not fit for cases when direct access to a certain data record is required. Furthermore, due to offline batch orientation of Hadoop it is not suited for low-latency applications.

GridGain In-Memory Compute Grid (IMCG) on the other hand is geared towards in-memory computations and very low latencies. GridGain IMCG has its own implementation of MapReduce which is designed specifically for real-time in-memory processing use cases and is very different from Hadoop one. Its main goal is to split a task into multiple sub-tasks, load balance those sub-tasks among available cluster nodes, execute them in parallel, then aggregate the results from those sub-tasks and return them to user.

Splitting tasks into multiple sub-tasks and assigning them to nodes is the mapping step and aggregating of results is reducing step. However, there is no concept of mandatory data built in into this design and it can work in the absence of any data at all which makes it a good fit for both, stateless and state-full computations, like traditional HPC. In cases when data is present, GridGain IMCG will also automatically colocate computations with the nodes where the data is to avoid redundant data movement.

It is also worth mentioning, that unlike Hadoop, GridGain IMCG is very well suited for processing of computations which are very short-lived in nature, e.g. below 100 milliseconds and may not require any mapping or reducing.

Here is a simple Java coding example of GridGain IMCG which counts number of letters in a phrase by splitting it into multiple words, assigning each word to a sub-task for parallel remote execution in the map step, and then adding all lengths receives from remote jobs in reduce step.

    int letterCount = g.reduce(
        BALANCE,
        // Mapper
        new GridClosure<String, Integer>() {
            @Override public Integer apply(String s) {
                return s.length();
            }
        },
        Arrays.asList("GridGain Letter Count".split(" ")),
        // Reducer
        F.sumIntReducer()
    ));

GridGain In-Memory Data Grid vs Hadoop Distributed File System

Hadoop Distributed File System (HDFS) is designed for storing large amounts of data in files on disk. Just like any file system, the data is mostly stored in textual or binary formats. To find a single record inside an HDFS file requires a file scan. Also, being distributed in nature, to update a single record within a file in HDFS requires copying of a whole file (file in HDFS can only be appended). This makes HDFS well-suited for cases when data is appended at the end of a file, but not well suited for cases when data needs to be located and/or updated in the middle of a file. With indexing technologies, like HBase or Impala, data access becomes somewhat easier because keys can be indexed, but not being able to index into values (*secondary indexes*) only allow for primitive query execution.

GridGain In-Memory Data Grid (IMDG) on the other hand is an in-memory key-value data store. The roots of IMDGs came from distributed caching, however GridGain IMDG also adds transactions, data partitioning, and SQL querying to cached data. The main difference with HDFS (or Hadoop ecosystem overall) is the ability to transact and update any data directly in real time. This makes GridGain IMDG well suited for working on operational data sets, the data sets that are currently being updated and queried, while HDFS is suited for working on historical data which is constant and will never change.

Unlike a file system, GridGain IMDG works with user domain model by directly caching user application objects. Objects are accessed and updated by key which allows IMDG to work with volatile data which requires direct key-based access.

GridGain IMDG allows for indexing into keys and values (i.e. primary and secondary indices) and supports native SQL for data querying & processing. One of unique features of GridGain IMDG is support for distributed joins which allow to execute complex SQL queries on the data in-memory without limitations.

GridGain and Hadoop Working Together

To summarize:

Hadoop essentially is a Big Data warehouse which is good for batch processing of historic data that never changes, while GridGain, on the other hand, is an In-Memory Data Platform which works with your current operational data set in transactional fashion with very low latencies. Focusing on very different use cases make GridGain and Hadoop very complementary with each other.

Up-Stream Integration

The diagram above shows integration between GridGain and Hadoop. Here we have GridGain In-Memory Compute Grid and Data Grid working directly in real-time with user application by partitioning and caching data within data grid, and executing in-memory computations and SQL queries on it. Every so often, when data becomes historic, it is snapshotted into HDFS where it can be analyzed using Hadoop MapReduce and analytical tools from Hadoop eco-system.

Down-Stream Integration

Another possible way to integrate would be for cases when data is already stored in HDFS but needs to be loaded into IMDG for faster in-memory processing. For cases like that GridGain provides fast loading mechanisms from HDFS into GridGain IMDG where it can be further analyzed using GridGain in-memory Map Reduce and indexed SQL queries.

Conclusion

Integration between an in-memory data platform like GridGain and disk based data platform like Hadoop allows businesses to get valuable insights into the whole data set at once, including volatile operational data set cached in memory, as well as historic data set stored in Hadoop. This essentially eliminates any gaps in processing time caused by Extract-Transfer-Load (ETL) process of copying data from operational system of records, like standard databases, into historic data warehouses like Hadoop. Now data can be analyzed and processed at any point of its lifecycle, from the moment when it gets into the system up until it gets put away into a warehouse.

GridGain 4.3.1 Released

2012-11-12T18:26:00.000-08:00

GridGain 4.3.1 service release includes several important bug fixes and host of new optimizations. It is 100% backward compatible and it is highly recommended update for anyone running production systems on 4.x code line.

Details

Date	November 10^th, 2012
Version	4.3.1e
Build	10112012

New Features and Enhancements

Added remove operation to data loader
Significantly improved performance of partition to node mapping
Added GridSerializationBenchmark for comparing performance of Java, Kryo, and GridGain serialization
Added property-based configuration to remote clients
Optimized concurrency for asynchronous methods in C++ client
Removed support for Groovy++ DSL Grover

Core Bug Fixes

Unmarshalling of SimpleDateFormat fails with NPE
Possible NPE in Indexing Manager when using distributed data structures
Swap partition iterator skips entries if off-heap iterator is empty
`GridDataLoader` does not allow to cache primitive arrays
Excessive memory consumption in indexing SPI
Add check on startup that GridOptimizedMarshaller is supported by running JDK version
If ordered message is timed out, other messages for the same topic may not be processed
ScalarPiCalculationExample does not provide correct estimate for PI

Client Connectivity Bug Fixes

Client router with explicit default configuration leads to NPE.
Repair REST client support to make session token and client ID optional
Ping does not work properly in C++ client

Visor Management Bug Fixes

Clear and Compact operations in Visor do not account for node selection
Move Visor management tasks into a separate thread pool
Preload dialog in Visor does not show correct number of keys
GC dialog in Visor waits indefinitely for dead nodes
Increase tooltip dismiss time in Visor
Visor log search does not show nodes table correctly on Windows

Download it now

In-Memory Data Grids... Explained

2012-11-01T21:46:00.000-07:00

In-memory processing has been a pretty hot topic lately. Many companies that historically would not have considered using in-memory technology because it was cost prohibitive are now changing their core systems’ architectures to take advantage of the low-latency transaction processing that in-memory technology offers. This is a consequence of the fact that the price of RAM is dropping significantly and rapidly and as a result, it has become economical to load the entire operational dataset into memory with performance improvements of over 1000x faster. In-Memory Compute and Data Grids provide the core capabilities of an in-memory architecture.

The goal of In-Memory Data Grids (IMDG) is to provide extremely high availability of data by keeping it in memory and in highly distributed (i.e. parallelized) fashion. By loading Terabytes of data into memory IMDGs are able to support most of the Big Data processing requirements today.

At a very high level, an IMDG is a distributed object store similar in interface to a typical concurrent hash map. You store objects with keys. Unlike traditional systems where keys and values are often limited to byte arrays or strings, with IMDGs you can use any domain object as either value or key. This gives tremendous flexibility by allowing you to keep exactly the same object your business logic is dealing with in the Data Grid without the extra step of marshaling and de-marshaling that alternative technologies would require. It also simplifies the usage of your data grid as you can, in most cases, interface with the distributed data store as you do with a simple hash map. Being able to work with domain objects directly is one of the main differences between IMDGs and In-Memory Databases (IMDB). In the case of the latter, users still need to perform Object-To-Relational Mapping which typically adds significant performance overhead.

There are also some other features in IMDGs that distinguish them from other products, such as NoSql databases, IMDBs, or NewSql databases. One of the main differences is truly scalable Data Partitioning across the cluster. Essentially IMDGs in their purest form can be viewed as distributed hash maps with every key cached on a particular cluster node - the bigger the cluster, the more data you can cache. The trick to this architecture is to make sure that you collocate your processing with the cluster nodes where data is cached to make sure that all cache operations become local and that there is no (or minimal) data movement within the cluster. In fact, when using well designed IMDGs, there should be absolutely no data movement on stable topologies. The only time when some of the data is moved is when new nodes join in or some existing nodes leave the cluster, thus causing some data repartitioning within the cluster.

The picture below shows a classic IMDG with a key set of {k1, k2, k3} where each key belongs to a different node. The external database component is optional. If present, then IMDGs will usually automatically read data from the database or write data to it.

Another distinguishing characteristic of IMDGs is Transactional ACID support. Generally a 2-phase-commit (2PC) protocol is used to ensure data consistency within cluster. Different IMDGs will have different underlying locking mechanisms, but usually more advanced implementations will provide concurrent locking mechanisms (GridGain, for instance, uses MVCC - multi-version concurrency control) and reduce network chattiness to a minimum, hence guaranteeing transactional ACID consistency with very high performance.

Data consistency is one of the main differences between IMDGs and NoSQL databases. NoSQL databases are usually designed using an Eventual Consistency (EC) approach where data is allowed to be inconsistent for a period of time as long as it will become consistent *eventually*. Generally, the writes on EC-based systems are somewhat fast, but reads are slow (or to be more precise, as fast as writes are). Latest IMDGs with an *optimized* 2PC should at least match if not outperform EC-based systems on writes, and be significantly faster on reads. It is interesting to note that the industry has made a full circle moving from a then-slow 2PC approach to the EC approach, and now from EC to a much faster *optimized* 2PC.

Different products provide different 2PC optimizations, but generally the purpose of all optimizations is to increase concurrency, minimize network overhead, and reduce the number of locks a transaction requires to complete. As an example, Google's distributed global database, Spanner, is based on a transactional 2PC approach simply because 2PC provided a faster and more straightforward way to guarantee data consistency and high throughput compared to MapReduce or EC.

Even though IMDGs usually share some common basic functionality, there are many features and implementation details that are different between vendors. When evaluating an IMDG product pay attention to eviction policies, (pre)loading techniques, concurrent repartitioning, memory overhead, etc... Also pay attention to the ability to query data at runtime. Some IMDGs, such as GridGain for example, allow users to query in-memory data using standard SQL, including support for distributed joins, which is pretty rare.

Storing data in an IMDG is only half of the functionality required for an in-memory architecture. This data must also be processed in a high-performance and parallelized manner. The typical in-memory architecture partitions data across the cluster using an IMDG, and then computations are sent to the nodes where the data is for collocated (local) execution. Since computations are usually part of Compute Grids and have to be properly deployed, load-balanced, failed-over, or scheduled, the integration between Compute Grids and IMDGs is very important. It is especially beneficial if both In-Memory Compute and Data Grids are part of the same product and utilize the same APIs which removes the burden of integration from the developer and usually renders the highest performance and reliable in-memory systems.

IMDGs (together with Compute Grids) are used throughout a wide spectrum of industries in applications as diverse as Risk Analytics, Trading Systems, real time Fraud Detection, Biometrics, eCommerce, or Online Gaming. Essentially every project that struggles with scalability and performance can benefit from In-Memory Processing and IMDG architecture.

MergeSort on GridGain

2012-09-21T14:38:00.001-07:00

Here is an example of how you can perform MergeSort on a distributed grid product like GridGain. This example is somewhat artificial, as you probably would never do the same thing in real life (executing the same code locally is most likely faster), but it does demonstrate some pretty cool features of GridGain, like recursive task execution and continuations.

This is the task class which splits array in two and sends remote jobs to sort the new arrays. Remote jobs in their turn execute the same task over and over again until we get to array size of 1, after which we begin merge process.

class GridMergeSortTask extends GridTaskSplitAdapter<int[], int[]> {
    // Injected Grid instance.
    @GridInstanceResource private Grid grid;

    @Override 
    protected Collection<GridJob> split(int gridSize, int[] initArr) {
        Collection<GridJob> jobs = new LinkedList<GridJob>();

        for (final int[] arr : splitArray(initArr)) {
            jobs.add(new GridJobAdapterEx() {
                // Auto-inject job context.
                @GridJobContextResource
                private GridJobContext jobCtx;

                // Task execution result future.
                private GridTaskFuture<int[]> fut;

                @Override public Object execute() throws GridException {
                    if (arr.length == 1)
                        return arr;

                    // Future is null before holdcc() is called and
                    // not null after callcc() is called.
                    if (fut == null) {
                        // Launch the recursive child task asynchronously.
                        fut = grid.execute(new GridMergeSortTask(), arr);

                        // Add a listener to the future, that will resume the
                        // parent task once the child one is completed.
                        fut.listenAsync(new GridInClosure<GridFuture<int[]>>() {
                            @Override public void apply(GridFuture<int[]> fut) {
                                // CONTINUATION:
                                // =============
                                // Resume suspended job execution.
                                jobCtx.callcc();
                            }
                        });

                        // CONTINUATION:
                        // =============
                        // Suspend job execution to be continued later and
                        // release the executing thread.
                        return jobCtx.holdcc();
                    }
                    else {
                        assert fut.isDone();

                        // Return the result of a completed child task.
                        return fut.get();
                    }
                }
            });
        }

        return jobs;
    }

    /**
     * GridTask reduce logic. This method is called when both child jobs
     * are completed, and is a Reduce step of Merge Sort algorithm.
     */
    @Override public int[] reduce(List<GridJobResult> results) {
        // This is in case we have a single-element array.
        if (results.size() == 1)
            return results.get(0).getData();

        assert results.size() == 2;

        int[] arr1 = results.get(0).getData();
        int[] arr2 = results.get(1).getData();

        return mergeArrays(arr1, arr2);
    }

    private static Iterable<int[]> splitArray(int[] arr) {
        int len1 = arr.length / 2;
        int len2 = len1 + arr.length % 2;

        int[] a1 = new int[len1];
        int[] a2 = new int[len2];

        System.arraycopy(arr, 0, a1, 0, len1);
        System.arraycopy(arr, len1, a2, 0, len2);

        System.out.println("Split array [arr1Len=" + a1.length + 
            ", arr2Len=" + a2.length + ']');

        return Arrays.asList(a1, a2);
    }

    private static int[] mergeArrays(int[] arr1, int[] arr2) {
        int[] ret = new int[arr1.length + arr2.length];

        int i1 = 0;
        int i2 = 0;

        // Merge 2 arrays into a resulting array
        for (int i = 0; i < ret.length; i++) {
            if (i1 >= arr1.length) {
                System.arraycopy(arr2, i2, ret, i, arr2.length - i2);

                break;
            }
            else if (i2 >= arr2.length) {
                System.arraycopy(arr1, i1, ret, i, arr1.length - i1); 

                break;
            }
            else
                ret[i] = arr1[i1] <= arr2[i2] ? arr1[i1++] : arr2[i2++];
        }

        System.out.println("Merged arrays [resLen=" + ret.length + 
            ", arr1Len=" + arr1.length + ", arr2Len=" + arr2.length + ']');

        return ret;
    }
}

And here is how you would call this task:

    public static void main(String[] args) throws GridException {
        Grid grid = G.start();

        try {
            int[] inArr = generateRandomArray(30);

            System.out.println("Unsorted array: " + Arrays.toString(inArr));

            int[] outArr = grid.execute(new GridMergeSortTask(), inArr).get();

            System.out.println("Sorted array: " + Arrays.toString(outArr));
        }
        finally {
            G.stop(true);
        }
    }

    private static int[] generateRandomArray(int size) {
        int[] ret = new int[size];

        Random rnd = new Random();

        for (int i = 0; i < ret.length; i++)
            ret[i] = rnd.nextInt(100);

        return ret;
    }

Enjoy!

10 Useful Performance Tuning Tips For GridGain

2012-09-05T14:15:00.002-07:00

I have been getting many questions of how to tune GridGain, so I decided to create a brief manual which covers most important tuning properties.

1. GridGain is multi-threaded - Use It

If you are experiencing somewhat slow performance for cache updates, you should ask yourselves whether you are utilizing full computing power (all the cores) on your machine. GridGain is multi-threaded internally, but if you are doing sequential operations one after another from a single thread, then you are not using multithreading. Generally it makes sense to use the amount of threads of about 2 or 3 times the number of cores for populating grid. All GridGain APIs are thread-safe, so you don't have to worry about any concurrency issue when populating data.

2. Use Collocated Computations

GridGain enables you to execute MapReduce computations in memory. However, most computations usually work on some data which is cached on remote grid nodes. Loading that data from remote nodes is usually expensive and it is a lot more cheaper to send the computation to the node where the data is. The easiest way to do it is to use GridProjection.affinityRun(...) method; however GridGain has plenty of "mapKeysToNodes(...)" methods to help users figure out data ownership within Grid.

3. Use Data Loader

If you need to upload lots of data into cache, use org.gridgain.grid.GridDataLoader to do it. Data loader will properly batch the updates prior to sending them to remote nodes and will properly control number of parallel operations taking place on each node to avoid thrashing. Generally it provides performance of 10x than doing a bunch of single-threaded updates.

4. Tune Initial Cache Size

To avoid internal resizing of cache maps you should always provide proper cache start size - not doing so can significantly hurt performance as some CPU cycles will be spent on GridGain resizing internal cache maps instead of application logic. You can configure cache start size via GridCacheConfiguration.getStartSize() configuration property.

5. Tune Near Cache

When using Partitioned cache, GridGain will front this cache with local Near cache to make sure that if entry does not belong to local partitions, it will still be cached in a smaller local cache for better performance on next access.

However, most usages of GridGain happen from collocated computations, i.e. computations submitted to the grid are usually routed to the nodes where the data resides automatically. In cases like this, using Near cache is redundant, as all data access happens from memory anyway. To save on performance, you can disable Near cache by setting GridCacheConfiguration.isNearEnabled() configuration property.

6. Tune Off-Heap Memory

If you plan to allocate large amounts of memory to your JVM for data caching (usually more than 10GB of memory), then your application will most likely suffer from prolonged lock-the-world GC pauses which can significantly hurt latencies. To avoid GC pauses use off-heap memory to cache data - essentially your data is still cached in memory, but JVM does not know about it and GC is not affected.

The only configuration property to set to enable off-heap memory is GridCacheConfiguration.getMaxOffHeapMemory() which will tell GridGain how much off-heap memory to make available for your application. By default off-heap memory is disabled.

7. Tune Swap Storage

First of all, if you don't plan to use swap storage (i.e. disk overflow storage), you should not change any default swap settings (swap storage is disabled by default). If you do need to use swap storage, then you should enable it via GridCacheConfiguration.isSwapEnabled() configuration property.

8. Tune Query Indexing

There are several configuration properties that you should watch out for here. First of all and most importantly, if you don't plan to use cache queries at all, you should disable indexing altogether via GridCacheConfiguration.isQueryIndexEnabled() configuration property.

If you do plan to use cache queries, you should properly enable/disable indexing of primitive keys and values on GridH2IndexingSpi. You should enable indexing for primitive keys by setting setDefaultIndexPrimitiveKey() to true on the SPI only if you plan to use primitive cache keys in your cache queries. The same goes for indexing primitive values controlled by setDefaultIndexPrimitiveValue(...) property.

Also, if for every value class you don't plan to have different key classes (essentially every value class has one key class), set setDefaultIndexFixedTyping(...) on the SPI to true. This way GridGain will store key types as corresponding SQL types instead of binary form which provides faster performance for key lookups.

9. Tune Eviction Policy

Again, if you don't plan to over-populate your cache, i.e. if you don't need any eviction policy at all, then you should disable eviction policy altogether via GridCacheConfiguration.isEvictionEnabled() configuration property.

If you do need GridGain to make sure that data in cache does not overgrow beyond allowed memory limits, you should carefully choose the eviction policy you need. Most likely you will need either FIFO or LRU eviction policies shipped with GridGain, however depending on your application, you may need to configure LIRs or plugin your own custom eviction policy. Regardless of which eviction policy you use, you should carefully chose the maximum amount of entries in cache allowed by eviction policy - if cache size overgrows this limit, then evictions will start occurring. Usually max size is controlled by setMaxSize(...) configuration property on the instance of eviction policy.

You should also almost always configure "setAllowEmptyEntries(...)" configuration property to false. By default GridGain will keep entries with null values in cache to preserve some other properties of the entry, like time-to-live for example. However, if you don't use time-to-live then most likely you should discard the entry once it gets expired or invalidated.

10. Use Write-Behind Caching

If you can afford for your persistent store to be behind your in-memory cache, then use write-behind caching. When write-behind is enabled, GridGain will batch up cache updates and flush them to database in batches in the background which can often provide significant performance benefits. You can enable write-behind caching via GridCacheConfiguration.isWriteBehindEnabled() configuration property.

Can Delaying Events Boost Performance Of Your Cluster?

2012-07-18T20:19:00.002-07:00

Should you ever delay events in your cluster? Kind of interesting question. As engineers we are used to think that performance matters and delaying things goes directly against performance, right? Well, not always - there are quite a few cases in distributed environment when delaying things actually helps performance. For example, the most usual case is when you have a continuous stream of asynchronous computations on the grid. Since sending a computation is much faster than processing it, you can essentially overload grid and some sort of back pressure mechanism will be needed (usually you would just limit amount of active computations your grid can handle concurrently). However, in this blog I want to focus on delaying data operations rather than computations.

If you have heard about partitioned caching, then you perhaps know that whenever a new node joins the grid or an existing node leaves the grid, cluster repartitioning happens. This basically means that, in case of new node, it has to take responsibility for some of the data cached on other nodes, and in case of node leaving the grid, other nodes have to take responsibility for the data cached on that node. Essentially this results in data movement between data grid nodes. Picture below illustrates how keys get partitioned among caching data nodes (share-nothing-architecture):

Now imagine that you need to bring multiple nodes up concurrently. The 1st node that comes up will take responsibility for some portion of the data cached on other nodes and will start loading that portion of the data from other nodes. When a 2nd node comes up, it will also take responsibility for some portion of the data, including some data from the 1st node that was just started, and now portion of the data that was moved to 1st node will have to be moved to the 2nd node. Ouch - wouldn't it be more efficient to wait till 2nd node comes up to start data preloading? The same happens when nodes 3, 4, etc... come up. So the most efficient way to do preloading of keys and to avoid extra network traffic causes by moving data between newly started nodes is to delay preloading until the last node starts.

At GridGain, we introduced this Delayed Preloading feature in our latest 4.2 release and now user is in full control of when data preloading happens - either right away upon node start, after a certain delay, or manually from our Visor Devops Console (note the Preload button on the right hand side):

As you see it does help to delay certain events in distributed systems to achieve better performance.

Segmenting Locks for Concurrency

2012-06-28T14:06:00.000-07:00

A pretty cool locking mechanism which we use every now and then at GridGain is concurrent segmented "striped" locks. Sometimes, your objects are constantly recreated, so you can't really attach a mutex to them. For cases like those you can partition the object space into multiple segments (often by object hashcode value) and only acquire the lock on the segment an object belongs to. This way the more segments you have, the more concurrency you have.

Here is a code for our StripedLock that I thought would be good to share:

public class StripedLock {
    // Array of underlying locks.
    private final Lock[] locks;

    public StripedLock(int concurrencyLevel) {
        locks = new Lock[concurrencyLevel];

        for (int i = 0; i < concurrencyLevel; i++)
            locks[i] = new ReentrantLock();
    }

    public int concurrencyLevel() {
        return locks.length;
    }

    public Lock getLock(int key) {
        return locks[abs(key) % locks.length];
    }

    public Lock getLock(long key) {
        return locks[abs((int)(key % locks.length))];
    }

    // Account for Math.abs() returning negative value.
    private int abs(int key) {
        return key == Integer.MIN_VALUE ? 0 : Math.abs(key);
    }

    public Lock getLock(@Nullable Object o) {
        return o == null ? locks[0] : getLock(o.hashCode());
    }

    public void lock(int key) {
        getLock(key).lock();
    }

    public void lock(long key) {
        getLock(key).lock();
    }

    public void unlock(int key) {
        getLock(key).unlock();
    }

    public void unlock(long key) {
        getLock(key).unlock();
    }

    public void lock(@Nullable Object o) {
        getLock(o).lock();
    }

    public void unlock(@Nullable Object o) {
        getLock(o).unlock();
    }
}

Simple - yet powerful! Hope you find it useful.

HBase As Your Persistent Store?

2012-06-28T02:16:00.000-07:00

We have added many cool features in GridGain 4.1. One of them is tight integration with Hadoop ecosystem. There are two ways you can integrate with Hadoop. One way is upstream integration in which you efficiently load data from HDFS into In-Memory Cache (aka Data Grid) where it gets indexed for low-latency query access. Another way is downstream integration where HDFS is used as a persistent store and data gets flushed into it periodically from In-Memory Cache using write-behind cache feature.

With downstream integration users are able to hold the latest data in memory and use HDFS as historical data warehouse without any extra ETL process and without lags in data. Business applications can run queries and get instant analytical feedback on the whole data set which includes most recent data held in-memory and HDFS-based warehoused data, hence not loosing any data in query results at all.

Here is an example of how HDFS-based cache store would look like in this case. I show how HBase would be used, but you can write directly to HDFS if you like. Note that all we do here is override methods load(...), put(...), and remove(...) to tell GridGain how to load and update entries in HBase.

Full source of this example can also be found on GitHub GridGain Project:

public class GridCacheHBasePersonStore 
    extends GridCacheStoreAdapter<Long, Person> {
    // Default config path.
    private static final String CONFIG_PATH "/my/hbase/hbase-site.xml";

    // Table name.
    private static final String TABLE_NAME = "persons";

    // Maximum allowed pool size.
    private static final int MAX_POOL_SIZE = 4;

    // HBase table pool.
    private HTablePool tblPool;

    // HBase column descriptor for first name.
    private HColumnDescriptor first = new HColumnDescriptor("firstName");

    // HBase column descriptor for last name.
    private HColumnDescriptor last = new HColumnDescriptor("lastName");

    public GridCacheHBasePersonStore() throws Exception {
        prepareDb();
    }

    // Load entry from HBase.
    @Override 
    public Person load(String cacheName, GridCacheTx tx, Long key) 
        throws GridException {
        HTableInterface t = tblPool.getTable(TABLE_NAME);

        try {
            Result r = t.get(new Get(Bytes.toBytes(key)));

            if (r == null)
                throw new GridException("Failed to load key: " + key);

            if (r.isEmpty())
                return null;

            Person p = new Person();

            p.setId(Bytes.toLong(r.getRow()));
            p.setFirst(Bytes.toString(r.getValue(first.getName(), null)));
            p.setLast(Bytes.toString(r.getValue(last.getName(), null)));
            
            return p;
        }
        catch (IOException e) {
            throw new GridException(e);
        }
        finally {
            close(t);
        }
    }

    // Store entry in HBase.
    @Override 
    public void put(String cacheName, GridCacheTx tx, Long key, Person val)
        throws GridException {
        HTableInterface t = tblPool.getTable(TABLE_NAME);

        try {
            t.put(new Put(Bytes.toBytes(key))
                .add(first.getName(), null, Bytes.toBytes(val.getFirst()))
                .add(last.getName(), null, Bytes.toBytes(val.getLast()));
        }
        catch (IOException e) {
            throw new GridException(e);
        }
        finally {
            close(t);
        }
    }

    // Remove entry from HBase.
    @Override 
    public void remove(String cacheName, GridCacheTx tx, Long key) 
        throws GridException {
        HTableInterface t = tblPool.getTable(TABLE_NAME);

        try {
            t.delete(new Delete(Bytes.toBytes(key)));
        }
        catch (IOException e) {
            throw new GridException(e);
        }
        finally {
            close(t);
        }
    }

    // Initialize HBase database.
    private void prepareDb() throws IOException {
        Configuration cfg = new Configuration();

        cfg.addResource(CONFIG_PATH);

        HBaseAdmin admin = new HBaseAdmin(cfg);

        if (!admin.tableExists(TABLE_NAME)) {
            HTableDescriptor desc = new HTableDescriptor(TABLE_NAME);

            desc.addFamily(first);
            desc.addFamily(last);

            admin.createTable(desc);
        }

        tblPool = new HTablePool(cfg, MAX_POOL_SIZE);
    }

    // Close HBase Table.
    private void close(@Nullable HTableInterface t) {
        ...
    }
}

To configure this store you should simply specify it in cache configuration and enable write-behind if you need data to be periodically flushed to HBase like so:

<bean class="org.gridgain.grid.cache.GridCacheConfigurationAdapter">
    ...
    <!-- Setup HBase Cache Store. -->
    <property name="store">
        <bean class="GridCacheHBasePersonStore" scope="singleton"/>
    </property>

    <!-- Enable write-behind. -->
    <property name="writeBehindEnabled" value="true"/>
    ...
</bean>

Adding LevelDB store for your In-Memory Cache?

2012-06-21T16:37:00.000-07:00

Recently, at one of the customer meetings, I was asked whether GridGain comes with its own database. Naturally my reaction was - why?!? GridGain easily integrates pretty much with any persistent store you wish, including any RDBMS, NoSql, or HDFS stores. However, then I thought, why not? We already have cache swap space (disk overflow) storage based on Google LevelDB key-value database implementation, so why not have the same for data store.

Here is how easy it was to add LevelDB based data store implementation for GridGain cache - literally took me 20 minutes to do, including unit tests. The store is based on GridGain swap space, but since swap space is based on LevelDB, you essentially get LevelDB local store for your cached data.

public class GridCacheSwapSpaceStore<K, V> 
    extends GridCacheStoreAdapter<K, V> {
    // Default class loader.
    private ClassLoader dfltLdr = getClass().getClassLoader();

    @GridInstanceResource
    private Grid g; // Auto-injected grid instance

    @Override 
    public V load(String cacheName, GridCacheTx tx, K key) 
        throws GridException {
        return g.readFromSwap(spaceName(cacheName), key, classLoader(key));
    }

    @Override 
    public void put(String cacheName, GridCacheTx tx, K key, V val) 
        throws GridException {
        g.writeToSwap(spaceName(cacheName), key, val, classLoader(val, key));
    }

    @Override 
    public void remove(String cacheName, GridCacheTx tx, K key) 
        throws GridException {
        g.removeFromSwap(spaceName(cacheName), key, null, classLoader(key));
    }

    private String spaceName(String cacheName) {
        return cacheName == null ? 
            "gg-spacestore-default" : "gg-spacestore-" + cacheName;
    }

    private ClassLoader classLoader(Object... objs) {
        ClassLoader ldr = null;

        for (Object o : objs) {
            if (o != null) {
                // Detect class loader for given object.
                ldr = U.detectClassLoader(o.getClass());

                if (ldr != dfltLdr)
                    break;
            }
        }

        return ldr;
    }
}

Quite easily done in my view. It will become part of next release of GridGain, so you will have local persistent store out-of-the-box if needed.

Plenty of more examples of different GridGain cache store implementations can be found on GitHub here.

Can You Manage Your Grid From Code?

2012-06-04T02:44:00.001-07:00

Most of the grid products on the market tout their GUI management consoles for managing and monitoring of their grids. Trying to use a grid without management is the same as trying to drive without GPS navigation - you can get from point A to point B, but it requires a lot more effort. For example, if you have more than a couple of nodes in your grid, how do you know which nodes are alive, what data they are caching, or what tasks they are executing? Management consoles would usually provide such functionality. At GridGain we have built a very feature-rich management and monitoring GUI tool called Visor (you can see screenshots here).

However, while being able to manage grid from GUI is important, it is equally as important to be able to do the same from code. What if you need to start or stop nodes programmatically when reacting to certain parameters of your business logic? Or what if you need to free up some more memory before executing a memory-intensive task? GridGain Remote Closure Execution in combination with Zero Deployment makes it very easy.

For example, here is how you would start or restart multiple remote nodes at once:

// By specifying startup parameters directly
G.grid().startNodes(
    hostSpec, // Mappings of host related parameters
    null, // Common defaults for this start routine. 
    false, // Restart existing nodes, if any.
    2000, // Timeout to wait for nodes startup. 
    5 // Max connections to each host for concurrent startup.
);

// Or by specifying all startup parameters in a file.
G.grid().startNodes(
    new File("/my/grid/startup/specification.ini", 
    false // Restart flag.
);

Or here is how you can compact your in-memory distributed cache on all grid nodes by freeing up internal byte buffers:

// Get all nodes on which "mycache" is enabled.
GridProjection gridPrj = G.grid().projectionForCache("mycache");

gridPrj.callAsync(new Callable<Object>() {
    public Object call() throws Exception {
        GridCache cache = G.grid().cache("mycache");

        // Compact all entries in cache.
        cache.compactAll(cache.keySet());

        return null;
    }
});

Or here is how you can free all memory used for cache entry backups on all grid nodes and move it onto disk storage:

// Get all nodes on which "mycache" is enabled.
GridProjection gridPrj = G.grid().projectionForCache("mycache");

gridPrj.callAsync(new Callable<Object>() {
    public Object call() throws Exception {
        GridCache cache = G.grid().cache("mycache");

        // Evict will move in-memory state onto swap storage on disk.
        // Note that we pass a predicate into evict method to make sure 
        // that only backup entries will be evicted.
        cache.evictAll(new GridPredicate<GridCacheEntry<Object, Object>>() {
            public boolean apply(GridCacheEntry<Object, Object> entry) {
                // Return true if entry is a backup of some other entry.
                return entry.backup();
            }
        });

        return null;
    }
});

This is just a small subset of a rich set of management hooks available in GridGain. In the above examples we have been utilizing the same remote closure execution as used for basic MapReduce tasks (we have more advanced API for more complex MapReduce tasks). A cool thing to note is that our closures created above will be deployed on the grid automatically using GridGain Zero Deployment functionality - tasks and closures never have to be predeployed on remote grid nodes.

What does "Real Time" in "Real Time Big Data" mean?

2012-05-13T19:22:00.001-07:00

Lately there has been lots of noise about "Real Time" Big Data. Lots of companies that associate themselves with this term are generally in analytical space and to them it really means "low-latency" for analytical processing of data which is usually stored in some warehouse, like Hadoop Distributed File System (HDFS). To achieve this they usually create in-memory indexes over HDFS data which allows their customers run fast queries on it.

Although low latencies are very important, they only cover one side of what "Real Time" really means. The part that is usually not covered is how current the analyzed data is, and in case of HDFS it is as current as the last snapshot copied into it. The need for snapshotting comes from the fact that most businesses are still running on traditional RDBMS systems (with NoSql gaining momentum where it fits), and data has to be at some point migrated into HDFS in order to be processed. Such snapshotting is currently part of most Hadoop deployments and it usually happens once or twice a day.

So how can your business run in "Real Time" if most of the decisions are actually made based on yesterday's data? The architecture needs to be augmented to work with LIVE data, the data which has just been changed or created - not the data that is days old. This is where In-Memory Data Grids (a.k.a. Distributed Partitioned Caches) come in.

By putting a Data Grid in front of HDFS we can store recent or more relevant state in memory which allows instant access and fast queries on it. When the data is properly partitioned, you can treat your whole Data Grid as one huge memory space - you can literally cache Terabytes of data in memory. But even in this case, the memory space is still limited and when the data becomes less relevant, or simply old, it should still be offloaded onto RDBMS, HDFS, or any other storage. However, with this architecture, businesses can now do processing of both, current and historic data - and this is very powerful. Now financial companies can quickly react to latest ticks in market data, gaming applications can react to latest player updates, businesses can analyze latest performance of ad campaigns, etc.

Here are some of the important benefits our customers get when deploying GridGain In-MemoryCompute and Data Grid in the above architecture:

Partitioning of data
Real Time MapReduce
Integration between Compute and Data Grids and ability to collocate your computations with data (a.k.a. data affinity)
In-Memory data indexing and ability to run complex SQL queries, including ability to execute "SQL Joins" between different types of data
Native affinity-aware clients for Java, .NET, C++, Android (with Objective-C client right around the corner)
Tight integration with various storage systems, including any type of RDBMS, Hadoop HDFS, HBase, etc...

Note that in this architecture In-Memory Data and/or Compute Grids really complement warehouse technologies, like HDFS - one is for in-memory processing of current data and another is for processing of historic data.

Distributed HPC with Scala and GridGain, June 5, NYC Scala Meetup

2012-05-11T17:08:00.000-07:00

We’ll be talking about using GridGain for highly distributed HPC programming with Scala at NYC Scala Meetup on June 5th.

As one of the main example we will be walking through a realtime word counting program with constantly changing text and will compare it with a Hadoop word counting example. You will see some cool features of GridGain such as Auto-Discovery, Streaming MapReduce, Zero Deployment, Distributed Data Partitioning, and In-Memory SQL Queries — all coded live in the presentation. Towards the end we will have an extensive Q&A session and cover some interesting real life use cases.

All detailes are here: http://www.meetup.com/ny-scala/events/63655442/

GridGain Drivers on Cruise Control

2012-04-25T02:12:00.001-07:00

Latest GridGain releases have had a big focus on remote grid clients, or as we like to call them remote drivers. We have recently added Java/Android and C#/.NET clients with C++ and Objective-C clients right around the corner. The goal of remote drivers is to provide ability to use Grid API from environments where it does not make sense to start a node and join the grid, but yet it is still needed to access objects from cache or execute a MapReduce task on remote grid.

The real trick when designing clients is to make sure that the API's are feature rich, but yet portable between different languages. We used Google Protocol Buffers to ensure portability. This way an object can be stored in cache using .NET C# client and retrieved, for example, using Java or C++ clients.

Another real important feature for remote drivers is data affinity, especially when working with partitioned caches. Partition caches have a concept of data ownership where every node owns a certain chunk of data. This way data can be split across many nodes and the whole grid becomes one shared RAM storage. However, such approach only works well with collocation of computations and data. All client requests to data should come exactly to the node where the data is cached - otherwise there will be lots of redundant data movement (a.k.a. data noise) which will have adverse effect on scalability.

At GridGain we took specific effort to make sure that all remote GridGain drivers have an embedded notion of affinity - this way you can store an object in cache using Java and then retrieve it using C# client which will know exactly where to go to get it.

Here is an example of how you would use GridGain remote Java or .NET C# clients for basic data and compute operations:

// Get all remote grid nodes
for (GridClientNode node: client.compute().nodes())
    System.out.println("External IP addresses: " + node.externalAddresses())

// Connect to partitioned cache.
GridClientData rmtCache = client.data("partitioned")

// Get and update 10 keys. All cache operations 
// will go exactly to the node where data is cached.
for (int key = 0; key < 10; key++) {
    // Get data.
    String val = rmtCache.get(key);

    // Update data if it is equal to the one we just got.
    rmtCache.cas(key, val + i, val);

    // Asynchronously remove data.
    rmtCache.removeAsync(key);
}

// Connect to remote grid.
GridClientCompute rmtCompute = client.compute();

// Execute map-reduce task on remote grid.
GridClientFuture fut = rmtCompute.executeAsync("MyTask", "myarg");

// Wait for completion.
fut.get();

...

using X = System.Console;

// Get all remote grid nodes.
foreach (IGridClientNode node in client.Compute().Nodes())
    X.WriteLine("External IP addresses: " + node.ExternalIpAddresses);

// Connect to partitioned cache.
IGridClientData rmtCache = client.Data("partitioned");

// Get and update 10 keys. All cache operations 
// will go exactly to the node where data is cached.
for (int key = 0; key < 10; key++) {
    // Get data.
    String val = rmtCache.GetItem(key);

    // Update data if it is equal to the one we just got.
    rmtCache.Cas(key, val + i, val);

    // Asynchronously remove data.
    rmtCache.RemoveAsync(key);
}

// Connect to remote grid.
IGridClientCompute rmtCompute = client.Compute();

// Execute map-reduce task on remote grid.
IGridClientFuture fut = rmtCompute.ExecuteAsync("MyTask", "myarg");

// Wait for completion.
fut.Result;

...

Note how almost identical the client API's for Java and C# look. Also note that both, C# and Java clients use key affinity and send data operations exactly to the nodes where the data is.

To try GridGain and it's new remote drivers, download GridGain 4.0.2 here.

GridGain 4.0.2 Released!

2012-04-13T16:06:00.000-07:00

GridGain 4.0.2 has been released today. This is a point release that includes several bug fixes as well as number of new features.

With 4.0.1 we are introducing native support for .NET with our C# Client. C# Client provides native .NET/C# APIs for accessing both GridGain’s In-Memory Data Grid and Compute Grid from outside of the GridGain topology context. Internally it’s deferring to the REST protocol. Check out examples on GitHub.

C# Client is one of many native clients we’ll be releasing shortly including Objective-C, C++, PHP, Scala, Ruby, and some others we’re already working on.

Improved Support for 32-bit and 64-bit Systems

We’ve modified our scripts for better out-of-the-box support for 32-bit and 64-bit systems. We’ve had several clients complaint that additional configuration properties were required and specifically GridGain Visor didn’t work fully on 32-bit system with default configuration. All these issues have been resolved.

Enhancements to GridGain Visor

We are continuing making rapid improvements to GridGain Visor that is part of GridGain Enterprise and OEM editions.

We’ve added ability to specific the time span for chart views:

We’ve added nice in-place filtering for events in Dashboard:

You can now double click on event and to see its details:

You can download GridGain 4.0.2 here.

Streaming MapReduce - GridGain Will Present at TriJug on April 16

2012-04-13T13:03:00.004-07:00

Come and see me talk about Streaming Map Reduce on GridGain and do some live coding of famous Hadoop’s example of counting popular words… but in Real Time context on GridGain. As always – live coding from scratch in Scala is never dull!

See you there.

Counting Words In Real Time On GridGain

2012-03-30T19:05:00.003-07:00

I have promised a while back to publish the code from live coding GridGain presentation I did at QCon London. Since presentation was in Scala, the code I will be posting here is in Scala, but I will post Java version within a couple of days as well.

First a brief intro. We all know Hadoop's counting words example which takes a file with words and then produces another file with number of occurrences next to each word. Hadoop does this example very well, however the main caveat with Hadoop's example is that it is not real time.

The counting words example I did at QCon actually counted words in real time. The program was split into two parts. First part is responsible for loading the words in real time into GridGain data grid, and the second part was querying the grid every 3 seconds to continuously print out top 10 words stored so far.

The example was done using 'Scalar' - GridGain DSL for Scala, but it could have been done In Java as well using GridGain Java APIs.

Continuously Populate Words In Real Time

Let's start by continuously loading data grid with new words. To do that, I downloaded several books in text format and started concurrently reading them from the populate(...) method, one thread per book. For every word read, I store it in cache, having the word itself as a key and number of current occurrences as a value. Also note how we let grid asynchronously update cache using asynchronous run while reading the next line from the book file (in reality you would most likely have more than one asynchronous job or have GridGain data loading functionality do it for you).

def populate(threadPool: CompletionService, dir: File) {
  val bookFileNames = dir.list()

  // For every book, start a new thread and start populating cache
  // with words and their counts.
  for (bookFileName <- bookFileNames) {
    threadPool.submit(new Callable {
      def call() = {
        val cache = grid$.cache[String, JInt]

        var fut: GridFuture[_] = null;

        Source.fromFile(new File(dir, name)).getLines().foreach(line => {
          line.split("[^a-zA-Z0-9]").foreach(word => {
            if (!word.isEmpty) {
              if (fut != null)
                fut.get()

              fut = grid$.affinityRunAsync(null, word, () => {
                // Increment word counter and store it in cache.
                // We use cache transaction to make sure that
                // gets and puts are consistent and atomic.
                cache.inTx(
                  () => cache += (word -> (cache.getOrElse(word, 0) + 1))
                )

                ()
              })
            }
          })
        })

        None // Return nothing.
      }
    })
  }

  // Wait for all threads to finish.
  books.foreach(_ => threadPool.take().get())
}

Distributed SQL Query

Now let's implement our distributed query against GridGain data grid which will run every 3 seconds. Note that we are using standard SQL syntax to query remote grid nodes. Interesting enough that GridGain data grid allows you to use SQL virtually without any limitations. You can use any native SQL function and even SQL JOINs between different classes. Here, for example, we are using SQL length(...) function to only query words greater than 3 letters long just to get rid of frequent short articles like "a" or "the" in our searches. We are also using desc keyword to sort word counts in descending order and limit keyword to limit our selection only to 10 words.

def queryPopularWords(cnt: Int) {
  // Type alias for sequences of strings (for readability only).
  type SSeq = Seq[String] 

  grid$.cache[String, JInt].sqlReduce(
    // PROJECTION (where to run):
    grid$.projectionForCaches(null),
    // SQL QUERY (what to run):
    "length(_key) > 3 order by _val desc limit " + cnt,
    // REMOTE REDUCER (how to reduce on remote nodes):
    (it: Iterable[(String, JInt)]) =>
      // Pre-reduce by converting 
      // Seq[(String, JInt)] to Map[JInt, Seq[String]].
      (it :\ Map.empty[JInt, SSeq])((e, m) => 
        m + (e._2 -> (m.getOrElse(e._2, Seq.empty[String]) :+ e._1))),
    // LOCAL REDUCER (how to finally reduce on local node):
    (it: Iterable[Map[JInt, SSeq]]) => {
      // Print 'cnt' of most popular words collected from all remote nodes.
      (new TreeMap()(implicitly[Ordering[JInt]].reverse) ++ it.flatten)
        .take(cnt).foreach(println _)

      println("------------") // Formatting.
    }
  )
}

Start Example

And finally let's implement our main(...) method our populate(...) and queryPopularWords(...) methods we just defined.

def main(args: Array[String]) {
  // Initialize book directory
  val bookDir = new File(BOOK_PATH);

  // Start GridGain with specified configuration file.
  scalar("examples/config/spring-cache-popularwords.xml") {
    // Create as many threads as we have book, so we can use
    // thread per book to load data grid concurrently.
    val threadPool = Executors.newFixedThreadPool(bookDir.list.length);

    val popWordsQryTimer = new Timer("words-query-worker");

    try {
      // Schedule word queries to run every 3 seconds.
      popWordsQryTimer.schedule(new TimerTask {
        def run() {
          queryPopularWords(10) // Query top 10 words from data grid.
        }
      }, 3000, 3000)

      // Populate cache with word counts.
      populate(new ExecutorCompletionService(threadPool), bookDir)

      // Force one more run to print final counts.
      queryPopularWords(POPULAR_WORDS_CNT)
    }
    finally {
      popWordsQryTimer.cancel() // Cancel timer.

      threadPool.shutdownNow() // Graceful shutdown.
    }
  }
}

To execute the example, start several GridGain stand-alone nodes using examples/config/spring-cache-popularwords.xml configuration file and then start the example we just created from IDE. You may wish to add more printouts for better visibility of what's happening.

This example is also shipped with GridGain 4.0 and also available in GridGain GitHub Repository.

GridGain 4.0 Released - Real Time Big Data Platform

2012-03-26T10:38:00.000-07:00

I’m pleased to announce that today we released GridGain 4.0 – latest edition of our platform for Real Time Big Data processing. I’m proud that our team set this final deadline almost 5 months ago and we were able to hit without a single delay.

I’m especially proud of this fact because of the enormous complexity of the development process involved in making software like GridGain – dozens of production clients, testing on serious massively distributed environments, set of new features, and the usual array of setbacks that we had to go through to get here.

Needless to say that we have also grown significantly as a business in the last 6 months including more than doubling our team headcount, rolling out new website, new branding, sales team, messaging, press and analysts relationships, investment, and the whole scope of other business activities.

But… GridGain System is an engineering company first and foremost and I’ll talk about technology in GridGain 4.0:

Among other enhancements this release includes

Visor Monitoring

One of the biggest additions in 4.0 release is our Visor - GUI-based Management & Monitoring Tool which is included int GridGain Enterprise and GridGain OEM editions. Visor provides you with GUI-based and scriptable environment for managing and monitoring GridGain distributed installations.

With Visor you can perform various nodes actions

View topology with metrics

View metrics for any logical subgrid

See comprehensive historical charts

View advanced grid-wide events

And lots of other cool stuff...

Affinity-Aware Clients

In GridGain 4.0 we are finally introducing native clients for various languages. The 4.0 release includes native Java and Android clients with rich APIs to support our Compute Grid and Data Grid connectivity. Our native .NET, C++, Groovy, and Scala clients are already in testing stage and will be coming out shortly as well. After that we will be adding Objective-C, Ruby, PHP, Python, and Node.js clients.

All of our clients natively support essentially the same APIs specifically adapted to a certain language. You can execute MapReduce tasks, perform bunch of data operations, like storing and retrieving values to/from remote caches, compare-and-set/replace/put-if-absent atomic operations, etc... You can also subscribe to topology updates and get very creative with partitioning remote data grid into logical subgrids.

But one of the coolest features in our clients is affinity-awareness. This basically means that when working with data grids, GridGain will automatically figure out on which node the data is stored and will route client requests to that node. Imagine the amount of network trips you can save by retrieving data directly from the node which is responsible for storing it (same goes for updates). This feature is available for all of our native clients, not only for Java, which makes GridGain into the only native cross -language distributed Real Time Big Data platform.

Memcached Binary Protocol Support

In GridGain 4.0 we significantly enhanced our REST support for HTTP(S) and added Binary protocol support as well. What's even cooler is that our binary protocol is fully Memcached-compliant. As a matter of fact, during our testing we have been connecting to GridGain using available open source Memcached clients and executing commands on GridGain data grid.

Having said that, GridGain 4.0 Binary connectivity protocol supports a lot more than Memcached does. Essentially we have taken Memcached protocol as our starting point and significantly enhanced it with our own commands and features. For example, you can configure security with proper authentication and secure sessions for remote clients, or you can execute MapReduce tasks, get remote node topology, etc...

Security

In GridGain 4.0 we added a notion of secure grids. Grids can now request for nodes to be authenticated prior to joining them into topology. Authentication implementation is fully pluggable through our SPI-based architecture and comes with several implementations out of the box, such as Passcode or JAAS-based authentication.

Additionally remote clients can also be required to authenticate themselves and once authenticated, they establish a secure session with the server.

Both, authentication and secure-session SPIs are available in GridGain enterprise edition only.

Data Loaders + Hadoop HDFS Support

We have also added support for efficient concurrent data loading for our data grid. There are plenty of ways to load data into data grids, including using basic cache APIs or our support for bulk-loading of data from data stores. Data loaders make it easy to externally load data into grid by adding collocation with data nodes, sending concurrent data loading jobs and properly controlling the amount of memory consumed by data loading process.

As a good use case, you can use data loader to preload data from Hadoop HDFS in order to process it in Real Time on GridGain. In fact GridGain 4.0 comes with HDFS data loading example, GridCacheHdfsLoaderExample, which reads data from HDFS and then uses GridGain data loader to load it into data grid.

1000+ Nodes Guaranteed Discovery

In this release we enhanced our TCP-based discovery protocol with enterprise-proven support for network segmentation and half-connected sockets. Our discovery protocol was tested on thousands of grid nodes on Amazon EC2 cloud by us and by our customers to make sure there are no cluster or data inconsistencies. This protocol is already running on customer sites in several deployments quite successfully.

LevelDB Swap Space Implementation

In GridGain 4.0 you can load terabytes of data into cache. GridGain will try to fit as much of the data in memory as possible - the more grid nodes you have, the more memory is available for caching data. However, if you have more data than fits into the whole memory of the grid, you can use LevelDB swap implementation which is based on Google LevelDB storage to swap infrequently used data to disk. We found that LevelDB can efficiently store large amounts of data with a fairly small disk footprint (using compression). We have also enhanced it with our swap eviction policy to prevent infinite disk growth.

Download GridGain 4.0 and try it out.

GridGain 3.6.0 Released!

2012-01-15T22:00:00.000-08:00

I am glad to announce that we have released GridGain 3.6.0 - Real Time Big Data platform that allows anyone easily develop, scale and manage compute and data intensive big data applications using integrated Compute and In-Memory Data Grids with Java, Scala and Groovy native APIs.

Some of the key new features and enhancements:

“Write From Behind Caching” was finally added to IMDGs
REST batch operation support introduced (putAll, removeAll)
Eviction filters added
Segmentation handling for half-connected network sockets
Improvements in GridGain Visor
New eviction performance optimizations
Significant IMDG preloading performance enhancements
New optimization for lock-free reads and low contention writes
Cumulative bug fixes and multitude of performance

You can download it here.

GridGain is hiring

2011-11-21T12:55:00.001-08:00

GridGain Systems is rapidly growing and we are hiring talented Java/Scala top-level engineers in both of our locations:
- Saint Petersburg, Russia
- Foster City, CA, USA

Come work for one of the hotest startups in the Silicon Valley! Cutting edge work and top level benefits and compensation. We also guarantee sleepless nights, unsolvable problems and insane debugging of complex distributed systems just in the first week

All the details are here.

GridGain 3.5 Released!

2011-11-11T00:53:00.001-08:00

We’ve just released GridGain 3.5 – our latest stable release of GridGain’s Real Time Big Data platform that allows anyone easily develop, scale and manage compute and data intensive JVM based applications using integrated Compute and In-Memory Data Grid with Java, Scala and Groovy native APIs.

Some of the key new features and enhancements:

Significant performance improvements throughout the product
Previously deprecated APIs have been removed
Improvements and bug fixes in GridGain Visor
Enhancements to GridProjection interface
Ability to programmatically start remote nodes in bulk
Customizable closure-based MapReduce
Enhancements in affinity and co-location
Bug fixes in eviction policies in Data Grid
Significant improvements and performance enhancements in Swap SPI

Download GridGain 3.5 today!

PDC Theorem – The Underlying Principle of Real-Time Distributed Processing

2011-10-21T00:45:00.000-07:00

Nikita, CEO at GridGain Systems, wrote a cool blog on PDC Theorem – The Underlying Principle of Real-Time Distributed Processing where he talks in some detail about important principles of distributed programming, such as real-time processing, data partitioning, and collocation of computations and data. Take a look at it - should be a good and interesting read.

How about distributed queues?

2011-08-29T16:39:00.000-07:00

Did you ever wish you could take a data structure you are familiar with and distribute it over grid? For example, why not take java.util.concurrent.BlockingDeque and add something to it on one node and poll it from another node? Or why not have a distributed primary key generator which would guarantee uniqueness on all nodes? Or how about a distributed java.util.concurrent.atomic.AtomicLong which can be updated and read from any node on the grid? GridGain gives you such capability. What GridGain did is actually take most of the data structures from java.util.concurrent framework and made sure they could be used in distributed fashion.

In this blog I want to show how flexible GridGain distributed queues are. On top implementing java.util.Collection interface and supporting different modes of operation, like collocated vs. non-collocated, or bounded vs. ubounded modes, you can actually control how elements are ordered within queues. GridGain supports FIFO, LIFO, and Priority based queues out of the box.

FIFO queues (first-in-first-out) are the most traditional queues where elements are added from the tail and polled form the queue head. LIFO queues (last-in-first-out) resemble more of stack features instead of queues. In LIFO queues elements are added and polled from the tail.

But the most interesting queue type is Priority queue where user can control the order of the elements. Priority queues order elements within the queue based on priority attribute specified by the user. Priority attribute of a queue element is annotated via @GridCacheQueuePriority annotation. Here is an example of how priority queue can be created and used.

public void priorityQueueExample() {
    Random rand = new Random();

    Grid grid = G.grid();

    // Initialize new unbounded collocated priority queue.
    GridCacheQueue<PriorityItem> queue = 
        grid.cache().queue("myqueue", PRIORITY);

    // Store 20 elements in queue with random priority.
    for (int i = 0; i < 20; i++) {
        int priority = rand.nextInt(20);

        queue.put(new PriorityItem(priority, "somedata-" + i));
    }

    PriorityItem item = null;

    int lastPriority = 0;

    do {
        item = queue.poll();

        // Ensure the elements are correctly ordered based on priority.
        assert lastPriority <= item.priority();

        lastPriority = item.priority();
    }
    while (item != null);
}

...

// Class defining sample queue element with its priority specified via
// @GridCacheQueuePriority annotation attached to priority field.
private static class PriorityItem implements Serializable {
    // Priority of queue item.
    @GridCacheQueuePriority
    private final int priority;

    private final String data;

    private SampleItem(int priority, String data) {
        this.priority = priority;
        this.data = data;
    }

    public int priority() {
        return priority;
    }
}

Clever Singletons on the Grid

2011-08-15T11:47:00.000-07:00

When working in distributed environment often you need to have a consistent local state per grid node that is reused between various job executions. For example, what if multiple jobs require database connection pool for their execution - how do they get this connection pool to be initialized once and then reused by all jobs running on the same grid node? Essentially you can think about it as a per-grid-node singleton service, but the idea is not limited to services only, it can be just a regular Java bean that holds some state to be shared by all jobs running on the same grid node.

In GridGain versions 2.x and earlier this approach was handled by using @GridUserResource annotation to annotate fields within GridTask or GridJob classes to specify singleton beans. However, this approach was dependent on GridDeploymentMode configuration and, for ISOLATED or PRIVATE deployment modes, resource could be initialized multiple times, once per GridTask. This forced users to use various hacks in their logic and generally was not very convenient to use.

Starting with GridGain 3.0, GridNodeLocal per-grid-node local storage was introduced. The name was borrowed from ThreadLocal class in Java, because just like ThreadLocal provides unique per-thread space in Java, GridNodeLocal provides unique per-grid-node space in GridGain. GridNodeLocal implements java.util.concurrent.ConcurrentMap interface and is absolutely lock-free. In fact, it simply extends java.util.concurrent.ConcurrentHashMap implementation and, therefore, inherits all the methods available there.

Here is an example of how GridNodeLocal could be used to create some user specific singleton service from a simple GridGain job:

final Grid grid = G.start(..);

// Execute runnable job on some remote grid node.
grid.run(BALANCE, new Runnable() {
  public void run() {
    GridNodeLocal<String, MySingleton> nodeLocal = grid.nodeLocal();

    // 1. Check if f someone already created our singleton service. 
    MySingleton service = nodeLocal.get("myservice");

    if (service == null) {
      // 2. Create new singleton service.
      MySingleton other = 
        nodeLocal.putIfAbsent("myservice", service = new MySingleton(..));

      if (other != null) 
        service = other;
    }

    // Perform operations with our singleton service.
    ...
  }
});

Even better AtomicInteger and AtomicLong in Java

2011-06-22T20:03:00.000-07:00

Last week I blogged about "Better AtomicInteger and AtomicLong in Java" and how I added it to GridGain 'lang.utils' package to be used within GridGain project and by our user base. Recently I made it even better. The standard JDK atomic 'compareAndSet(..)' operations always check for some 'expected' value and if it matches, then the set happens. What if you simply want to set a value if its greater or less than current value, or if it simply does not match the current value?

Again, code looks pretty simple:

public class MyAtomicInteger extends AtomicInteger {
  ... // Omitting methods from previous blog
  public boolean setIfGreater(int update) {         
     while (true) {             
          int cur = get();

          if (update > cur) {
              if (compareAndSet(cur, update))
                  return true;
          }
          else
              return false;
      }
  }

  public boolean setIfGreaterEquals(int update) {
      while (true) {
          int cur = get();

          if (update >= cur) {
              if (compareAndSet(cur, update))
                  return true;
          }
          else
              return false;
      }
  }

  public boolean setIfLess(int update) {
      while (true) {
          int cur = get();
          
          if (update < cur) {
              if (compareAndSet(cur, update))
                 return true;
          }
          else
              return false;
      }
  }      
  
  public boolean setIfLessEquals(int update) {
      while (true) {
          int cur = get();

          if (update <= cur) {
              if (compareAndSet(cur, update))
                  return true;
          }
          else
              return false;
      }
  }

  public boolean setIfNotEquals(int update) {
      while (true) {
        int cur = get();

        if (update != cur) {
            if (compareAndSet(cur, update))
                return true;
        }
        else
            return false;
      }
  }
}