GridGain - Real Time Big Data

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;
      }
  }
}

Better AtomicLong and AtomicInteger in Java

2011-06-18T15:27:00.000-07:00

I was always annoyed by the fact that AtomicLong.compareAndSet(...) method in Java only checked for exact equality. What if you need to check if the value is less than or greater than? So I finally just created my own AtomicLong and AtomicInteger classes and added them to GridGain utilities to be used by us and by all our users.

The code is very simple:


public class MyAtomicLong extends AtomicLong {
    public boolean greaterAndSet(long check, long update) {
        while (true) {
            long cur = get();

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

    public boolean greaterEqualsAndSet(long check, long update) {
        while (true) {
            long cur = get();

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

    public boolean lessAndSet(long check, long update) {
        while (true) {
            long cur = get();

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

    public boolean lessEqualsAndSet(long check, long update) {
        while (true) {
            long cur = get();

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

Based on the comments I received, I have also added predicate-based 'checkAndSet(...)' method:


    public boolean checkAndSet(GridPredicate<Long> p, long update) {
        while (true) {
            long cur = get();

            if (p.apply(cur)) {
                if (compareAndSet(cur, update))
                    return true;
            }
            else
                return false;
        }
    }

Of course in GridGain these classes are named GridAtomicInteger and GridAtomicLong, since we prefix all our classes with the word Grid.

Remote Actors vs. GridGain and Concurrency Unification

2011-06-09T11:34:00.000-07:00

We are migrating and finishing the upcoming GridGain’s book “Pragmatic Cloud Computing with GridGain using Java and Scala”. Here’s a short chapter that I’ve just finished on the subject of:

“Remote Actors vs. GridGain and Concurrency Unification”

Enjoy!

GridGain 3.1.0 Released

2011-06-01T13:23:00.000-07:00

We’ve just released GridGain 3.1.0 – our latest stable release of GridGain’s High Performance Cloud Computing platform that allows anyone easily develop, scale and manage compute and data intensive JVM based applications.

Some of the key new features and enhancements (combined with 3.0.9 announcement last week):

Improvements in stability and performance for Data Grid
New Distributed Data Structures:
- GridCacheAtomicLong
- GridCacheAtomicReference
- GridCacheAtomicSequence
- GridCacheAtomicStamped
- GridCacheQueue
Fully distributed Functional Programming framework for Java
Distributed Java closure execution
Improved AOP-based distributed execution
Improved Data Grid querying support for:
- Native SQL (including full joints support)
- Predicate-based Full Scan
- Lucene
- H2 text
Support for local and remote reduction and transformations in Data Grid queries
Improved Scala-based DSL for Compute and Data Grid:
- Enhanced distributed closure execution with zero deployment
- Ad-hoc Data Grid queries
- Interactive GridGain Scala REPL
- Work with Compute and Data Grid from Scala REPL
Dramatically enhanced GridGain Visor for distributed monitoring and management
Brand new TCP/IP guaranteed discovery SPI for large cloud-based deployments
New enterprise licensing mechanism with dynamic hot-upgradability
New optimized marshaling
Improved documentation and examples

GridGain 3.1.0 comes in two editions:

Free Community Edition, GPLv3 licensed
Enterprise Edition

Enjoy!

Presenting at Scala Days 2011

2011-06-01T13:20:00.001-07:00

GridGain will be presenting at Scala Days 2011 on June 2nd, 2011. The event is hosted by Stanford University. The presentation starts at Session 3: Parallelism track at 3:50pm.

If you are around – stop by to see a non-stop Scala distributed coding. Should be fun!

Cool GridGain Transaction Example

2011-05-31T11:17:00.000-07:00

I never planned to create an example like this, but one of the users in our community forums pushed me to do it. Basically the user was complaining that he gets some deadlock behavior on the grid when he reads and updates the same object from cache on multiple grid nodes in parallel. Naturally I got worried and created the following test, which passes by the way :)

In this example we basically store array of size N in cache and execute N jobs on the grid, each of which gets an array from cache and flips the array cell assigned to it from zero to one. As a result you start out with an array of all zeros and you end up with array of all ones. The cache is configured in PARTITIONED mode with synchronous commits. By default, every key stored in the cache is backed up on another node. Here is how the code looks like:


// Number of grids in the test.
private static final int GRID_COUNT = 4;

public void testTransactions() throws GridException {
    // Start multiple grid nodes in the same JVM.
    for (int i = 0; i < GRID_COUNT; i++) {
        GridConfigurationAdapter cfg = new GridConfigurationAdapter();

        // I am specifically omitting other configuration details. 
        cfg.setGridName("grid-" + i);

        G.start(cfg);
    }
 
    try {
        // Try job count N = 1000
        checkTransactions(1000);
    }
    finally {
        G.stopAll(true);
    }
}
 
private void checkTransactions(final int jobCnt) throws GridException {
    // Get frist grid node.
    Grid grid = G.grid("grid-0");
 
    final GridCache<String, int[]> cache = grid.cache();
 
    Collection<GridFuture<?>> futs = new LinkedList<GridFuture<?>>();
 
    for (int i = 0; i < jobCnt; i++) {
        // Run these jobs in parallel on all grid nodes.
        GridFuture<?> fut = grid.runAsync(
            GridClosureCallMode.BALANCE, 
            new GridInClosureX<Integer>() {
                @Override 
                public void applyx(Integer i) throws GridException {
                    GridCacheTx tx = 
                        cache.startTx(PESSIMISTIC, REPEATABLE_READ);

                    try {
                        int[] arr = cache.get("TestKey");
 
                        if (arr == null)
                            arr = new int[jobCnt];
 
                        arr[i] = 1;
 
                        cache.put("TestKey", arr);

                        tx.commit();
                    }
                    finally {
                        tx.end();
                    }
                }
            }, 
            i);
 
        futs.add(fut);
    }
 
    for (GridFuture<?> fut : futs)
        fut.get(); // Wait for completion.
 
    // Get array from cache.
    int[] arr = cache.get("TestKey");
 
    assert arr != null;
    assert jobCnt == arr.length;
 
    // Make sure that all array cells got flipped to 1.
    for (int i : arr)
        assert i == 1;
}

By the way, we have recently released GridGain 3.1.0, so give it a try.

GridGain 3.0.9 released

2011-05-19T17:05:00.001-07:00

Long anticipated GridGain 3.0.9 has been released today.

Some of the highlights of this release:

Massive improvements in stability and performance for Data Grid.
New distributed data structures:
Distributed Atomic
Distributed Queues
Much improved Data Grid querying support.
Improved Scalar DSL.
Dramatically enhanced monitoring & management via GridGain Visor.
Brand new TCP-based discovery SPI with support for large cloud-based deployments.
Improved documentation and examples.
New licensing support with dynamic hot-upgrade capabilities.
New optimized marshaling.
Go ahead and download GridGain 3.0.9.

Enjoy!

Sitting on the cloud with Nikita Ivanov

2011-05-02T11:35:00.000-07:00

Nikita Ivanov, CEO at GridGain gave a pretty cool interview on Cloud Computing and how GridGain fits in it. You can read the interview at Dzone: http://jetbrains.dzone.com/articles/sitting-cloud-nikita-ivanov.

Enjoy!

Asynchronous Functional APIs in Java

2011-04-20T16:53:00.000-07:00

The major limitation of java.util.concurrent.Future class is lack of a good way to listen for future completion asynchronously. Even though Java futures are asynchronous in nature and meant to eliminate synchronous waits, the only way to get a future result is to synchronously wait for future completion. In many cases this defeats the purpose and Java futures cannot be used.

To avoid this problem in GridGain we created our own future which allows for asynchronous listening for future completion. In addition to that we have also resolved the inconveniences of error handling and added 'duration()'method to tell users how long the future took to execute. Here is how you can add a listener to get notified, say for completion of putAsync(..) operation on distributed cache:


...
GridCache<Integer, String> cache = grid.cache();

// Asynchronously put a value into cache.
GridFuture<String> fut = cache.putAsync(1, "1");

// Pass in a closure which will be called whenever future completes.
fut.listenAsync(new GridInClosureX<GridFuture<String>>() {
    public void applyx(GridFuture<String> fut) throws GridException {
        V previousValue = fut.get();

        long duration = fut.duration();

        System.out.println("Finished cache put operation [prev=" + 
            previousValue +  ", duration=" + duration + "ms]"); 
    }
});
...

GridFuture is used for all asynchronous operations in GridGain, including both, compute and data grid features. Hope you find it useful.

Querying Cached Data - Paradigm Shift

2011-03-29T15:17:00.000-07:00

Why would you ever query cached data if you can query your persistent store, such as database? Well, the answer is the same as for accessing data by key from cache vs. getting it from database - for performance and scalability. However, querying cache is not exactly the same as querying your database - the main difference is that if cache only has a subset of data stored in database, then you will be only querying that subset, so query result will be reflecting only in-memory state. Does this matter? Depends on your application requirements and also depends on the amount of data you are able to store in cache.

With introduction of cloud computing and virtual instances, the amount of memory available to your grid on the cloud becomes virtually limitless. Adding nodes to your grid has become as simple as calling AWS API on EC2 whenever your application demands it. On top of it, if GridGain swap space is configured, all the data that cannot fit in memory on a single node will be overflown to disk. Also, your application may not even have that much data, or maybe querying cached data, which usually contains data that has been accessed relatively recently, is good enough. Thus in many cases querying cache is becoming to look more and more like querying your database.

Now that you made a decision in your project that you want to query cached data, the next question becomes how to cache query results. Most of us are familiar with Hibernate and it's support for 2nd Level Caching which also comes with Query Cache. The way query cache works in Hibernate is generally the way we are used to think of caching queried data. In a nutshell, a query is issued against the database and the results of the query are then stored in cache in a single collection. If you have multiple queries, then multiple collections containing query results are stored. Now if you ever update a single bean in Hibernate which can potentially affect the query result (pretty much any change to the queried tables), Hibernate is forced to invalidate (remove) the cached query results from cache and reload them on-demand next time. This significantly increases memory consumption, and frequent cache invalidations of query results perform horribly and do not scale at all. Even Hibernate itself discourages its users from using it. Here is the quote from Hibernate documenation:

... most queries do not benefit from caching of their results. So by default, individual queries are not cached even after enabling query caching ...

So, how does querying of cached data help? It helps by entirely removing the need for Query Result Cache altogether. SQL queries on your indexed cached data are executed in memory and perform very fast, so there is no more need to cache query results. Just run your SQL query on your cached data and get the results whenever you need them. However, it is important to note that without rich SQL support for cache queries, they will not be able to replace database queries within your project. In the example below, where Person relates to Company, if your cache does not support SQL joins, then you would not be able to find all people working for the same company, which may be quite limiting. Hense, it is extremely important to evaluate how rich the SQL support on a certain cache product before making a decision to query cached data.

In GridGain 3.0 the support for cache queries is virtually without any limitations. If you know SQL, you can run queries against cached data, including support for any type of joins, where clause keywords, order by, group by, etc... In addition to SQL queries, GridGain also supports text queries using Lucene or H2 TEXT underlying indexing. You can also run predicate-based FULL SCAN queries, which will iterate over all cache elements on remote nodes and will include only the ones that passed the predicate filter.

As an examples take a look at some of the GridGain 3.0 query examples below. Note the JDBC PreparedStatement syntax for passing arguments and the SQL join performed between Person and Organization classes. Also note how you can cherry-pick the set of nodes on which you would like to execute your query.


// Create query which selects salaries based on range for all employees
// that work for a certain company.
GridCacheQuery<Long, Person> qry = cache.createQuery(SQL, Person.class,
 "from Person, Organization where Person.orgId = Organization.id " +
  "and Organization.name = ? and Person.salary > ? and Person.salary <= ?");  

// Query all nodes to find all cached GridGain employees 
// with salaries less than 1000. 
qry.queryArguments("GridGain", 0, 1000).execute(grid);  

// Query only remote nodes to find all remotely cached GridGain employees 
// with salaries greater than 1000 and less than 2000. 
qry.queryArguments("GridGain", 1000, 2000).execute(grid.remoteProjection()); 

// Query local node only to find all locally cached GridGain employees 
// with salaries greater than 2000. 
qry.queryArguments("GridGain", 2000, Integer.MAX_VALUE).
    execute(grid.localNode());

Here is an examples of a text query which will scan all resumes for word "master":


// Will query fields annotated with @GridCacheQueryLuceneField annotation.
GridCacheQuery<Long, Person> mastersQry = 
    cache.createQuery(LUCENE, Person.class, "Master");

// Query all cache nodes.
mastersQry.execute(grid));

GridGain Presents at TriJug

2011-03-16T14:25:00.000-07:00

I will be speaking about GridGain and its Scala usage at TriJUG on March 21st, Raleigh, North Carolina. Topic is about “Cloud Computing with Scala and GridGain”. As always – plenty of live Scala coding. World’s shortest MapReduce app in 5 minutes in Scala as well as Data Grid access using Scala.

Hope to see you there!

GridGain User Group Started

2011-03-11T10:21:00.000-08:00

GridGain user group started - Join us!

GridGain at CodeMotion, Rome, Italy, March 5th

2011-03-01T00:25:00.000-08:00

I will be presenting at CodeMotion Conference on March 5th on subject "Distributed Programming with GridGain and Scala". As usual, expect very little theory and lots of cool live coding examples around GridGain Data Grid and Compute Grid features.

Hope to see you there!

Compute Grids - What To Expect?

2011-02-14T13:25:00.000-08:00

Recently I have been seeing a few new products branding themselves as compute grids while in reality they have very little to do with compute grids. While data grids are provided by quite a few vendors and people generally know what to expect from them, compute grids are not as common, and people may end up thinking that a compute grid is a simple RPC-like function execution on a remote node. As you may have guessed, I will show that to be untrue.

Let's take Cloudy Akka for example. While it is a wonderful product with great ideas borrowed from Erlang, it's not really a compute grid. What it provides is a convenient way to execute functions remotely, but it does not provide all the underlying plumbing that we have come to expect from compute grids.

Having worked on GridGain compute grid myself for over 5 years already (and then on data grid), and having studied quite a few others, here are some features that are minimally required in any product in my opinion, before it can claim itself as a compute grid.

Auto Discovery - all nodes in the grid should auto-discover each other, i. e. user should never have to manually add nodes to a topology.
MapReduce - support for splitting execution into multiple sub-jobs and then aggregating the results is just a MUST. Otherwise you are offloading most of the dirty work onto your users, which is not fair.
Auto Failure Detection - compute grid must be smart enough to automatically detect node crashes and proportionally distribute all the load among remaining nodes.
Fault Tolerance - all failed grid jobs must be automatically failed-over to other nodes, which are better suited for executing these jobs.
Load Balancing - compute grid should automatically distribute load equally among nodes, usually utilizing many different policies for load balancing. GridGain even has support for work-stealing, where less-loaded nodes can steal jobs from overloaded nodes.
Job Collision Resolution - this gives users control over how many jobs can run in parallel, while other jobs should wait in waiting queues, ordered by multiple available collision resolution strategies.
Auto Deployment - compute grid users should never be forced to manually deploy their libraries on all available grid nodes, this is just way too inconvenient and error-prone. The approach I like the best (available in GridGain) is auto-deployment, where code just automatically penetrates throughout the grid without any explicit action from users.
Nested Jobs And Continuations - compute grid jobs should be able to invoke other compute grid jobs when executing remotely. This is a very powerful feature, especially when grid jobs are recursive. Continuations should allow to suspend a job and release its resources while it's waiting for a result of another job within the grid.

I could probably continue, but I will stop here. I think it is clear that compute grid products are way more advanced than RPC frameworks. Just as a comparison, here are the GridGain examples implemented in GridGain Scalar which also do all the other cool stuff compute grids do in the background:


scalar {
    // 1. Execute (unicast) a simple job on some remote node.
    grid ucastRun (() => println("> GridGain ROCKS <"))

    // 2. Execute (broadcast) a simple job on all remote nodes.
    grid bcastRun (() => println("> GridGain ROCKS <"))      

    // 3. Use MapReduce to split a phrase into multiple words and     
    //    print each word on remote nodes.     
    grid splitRun 
        (for (w <- "GridGain ROCKS".split(" ")) yield () => println(w))

    // 4. Use MapReduce to count number characters by spreading
    //    workload to the grid and reducing on local node.
    val cnt = grid splitReduce 
        (for (w <- "GridGain REALLY ROCKS!".split(" "))
            yield () => w.length, // Map step.
            (s: Seq[Int]) => s.sum) // Reduce step.
}

Here is the similar example written in Java. It's a bit more verbose, but nevertheless, still pretty simple.


    // 1. Execute (unicast) a simple job on some remote node.
    grid.run(UNICAST, new GridRunnable() {
        @Override public void run() {
            System.out.println("> GridGain ROCKS <");
        }
    });

    // 2. Execute (broadcast) a simple job on all remote nodes.
    grid.run(BROADCAST, new GridAbsClosure() {
        @Override public void apply() {
            System.out.println("> GridGain ROCKS <");
        }
    });

    // 3. Use MapReduce to split a phrase into multiple words and     
    //    print each word on remote nodes.     
    grid.run(SPREAD, F.yield("> GridGain ROCKS <".split(" "), 
        new GridInClosure<String>() {
            @Override public void apply(String word) {
                System.out.println(word);
            }
    }));

    // 4. Use MapReduce to count number characters by spreading
    //    workload to the grid and reducing on local node.
    int cnt = grid.reduce(
        BALANCE,
        new GridClosure<String, Integer>() { // Create executable logic.
            @Override public Integer apply(String word) {
                return word.length();
            }
        },
        Arrays.asList("GridGain REALLY ROCKS!".split(" ")), // List of words.
        F.sumIntReducer() // Reducer which adds up all the values given to it.
    );

GridGain Scalar is available within GridGain release and can be downloaded here.

2011-02-04T14:53:00.000-08:00

Simon Forge at Ptak, Noel & Associates wrote a pretty nice article about "GridGain - An Open Source Success Story in Cloud Computing Land".

Have a look at it - it's not very technical, but contains quite useful market analysis.

Taste Of SQL For In-Memory Cache Queries

2011-01-28T09:58:00.000-08:00

I plan to blog in more detail about support of native SQL in GridGain data grid in a few days, but here is just a taste on how complex of an SQL statement you can execute against GridGain distributed in-memory cache. Note that you are not querying a database - you are querying objects cached in memory.

The query below utilizes H2 database SQL over JDBC syntax to query a collection of QueueItems at specified positions of a dynamic priority-based queue stored in distributed GridGain cache. As we know, querying an element at a certain position of a result set is not trivial in any SQL, and nested 'select' statements with 'rownum' field usually need to be used. On top of that we need to pass a collection of item positions to a query, which is also a fairly advanced feature.


String sql = 
    "select * from (" +
        "select *, rownum as r from (" +
            "select * from QueueItem where qid=? " + 
                "order by priority desc, seq asc" +
        ")" +
    ") where r in (select * from table(x int=?))"

GridCacheQuery itemsQry = cache.createQuery(SQL, QueueItem.class, sql);

// Specify positions of queue items we want to get.
Collection<Integer> positions = Collections.asList(10, 12, 14);

// Query items at specified positions from queue with ID 123.
GridCacheQueryFuture fut = itemsQry.queryArguments(123, positions).execute();

System.out.println("Queried queue items at specified positions: " + fut.get());

Pretty powerful in my view especially when you also get smart routing, pagination, custom transformers, reducers, visit-only support, text queries, and a lot more...

More information on GridGain queries can be found here or here.

Which Grid Product To Use?

2011-01-26T15:18:00.000-08:00

- Why use GridGain if there is Infinispan?
      - Why use Infinispan if there is Hazelcast?
            - Why use Hazelcast if there is Map over RMI?
                  - Why use Map over RMI if you can do it yourself?

I think you see the pattern... As you climb up this ladder you get more and more value from your grid product.

Collocation - First Rule Of Distributed Programming

2011-01-14T00:28:00.001-08:00

I am pleased to announce that we just recently released GridGain 3.0.4. The last couple of releases have been focused, among other things, around convenient and effective collocation of computations and data, and also grouping of data that is usually accessed together on the same nodes. Sending computations exactly to the nodes where the accessed data is residing is one of the key components in achieving better scalability. Without collocation, nodes fetch various data from other nodes for brief periods of time, just to perform often a quick computation and discard it almost immediately thereafter. This creates unnecessary data traffic, a.k.a. data noise, and can at times bring a server to its knees.

In my previous blog post I showed how to collocate computations and data using direct API via GridCache.mapKeyToNode(..) method. We have also added analogous methods on Grid API to provide capability of finding data affinity on the nodes that do not cache any data themselves. In our latest 3.0.4 release we have also added a very convenient way to provide collocation via @GridCacheAffinityMapped annotation.

Say you have 2 types of objects, Person and Company. Multiple persons can work for the same company. This means that you generally may wish to access Person objects together with the Company for which they work. To do that in a scalable fashion, you may wish to ensure that all people working for the same company are cached on the same node. This way you can send computations to that node and access multiple people from the same company locally. Here is how it can be done in GridGain.

public class PersonKey {
  // Person ID used to identify a person.
  private String personId;

  // Company ID which will be used for data affinity.
  @GridCacheAffinityMapped
  private String companyId;
  ...
}
...
// Instantiate person keys with same company ID.
Object personKey1 = new PersonKey("myPersonId1", "myCompanyId");
Object personKey2 = new PersonKey("myPersonId2", "myCompanyId");

// Both, the company and the person objects will be cached on the same node.
cache.put("myCompanyId", new Company(..));
cache.put(personKey1, new Person(..));
cache.put(personKey2, new Person(..));

Now, if you want to perform a computation which involves multiple people working for the same company, all you have to do is send a grid job to the node where those people are cached. Here is how you would send a computation to the node which caches all people for the company with ID "myCompanyId".

G.grid().run(GridClosureCallMode.BALANCE, new Runnable() {
  // This annotation specifies that computation should be routed
  // precisely to the node where all objects with affinity key
  // of 'myCompanyId' are cached.
  @GridCacheAffinityMapped
  private String companyId = "myCompanyId";

  @Override public void run() {
      // Some computation logic here.
      ...
  }
};

Now, when you properly collocate all your data within your data grid and then route your computations to the nodes where your data is cached, all cache operations become LOCAL, hence achieving best performance and scalability without any data noise. Kind of goes inline with the first rule of distributed programming, which is DO NOT DISTRIBUTE.

Say Hello To GridGain Data Grid

2011-01-05T19:33:00.000-08:00

I have been thinking how a HelloWorld example should look for data grid. After checking some other products I have noticed that the most popular approach for a HelloWolrd app on a data grid is creating an example which has two counter parts: client and server. The client part usually prints out the operation on cache, and the remote server would usually print out the same operation whenever the data ends up on that server. This way users can see that the value stored in cache actually does get distributed to remote nodes.

After looking at such examples it occurred to me that this client/server approach can be implemented a lot simpler in GridGain using zero deployment and basic event subscription. All we need to do is make sure that cache operations get printed out on remote nodes so we can visualize what's going on. However, for that we don't need to create a separate server app - we can do it all from our client example code.

So, let's make sure that events are printed out. For that we will execute a closure on all grid nodes which will subscribe to cache events and print them. This closure can be executed directly from example code and will be automatically deployed on remote nodes. Here is how the code will look like:

// Execute this runnable on all grid nodes, local and remote.
G.grid().run(BROADCAST, new Runnable() {
    @Override public void run() {
        // Event listener which will print out cache events, so we 
        // can visualize what happens on remote nodes.
        GridLocalEventListener lsnr = new GridLocalEventListener() {
            @Override public void onEvent(GridEvent e) {
                System.out.println("Event '" + e.type() + "' for key: " +
                    ((GridCacheEvent)e).key());
            }
        };

        // GridNodeLocal is a ConcurrentMap attached to every grid node.
        Object prev = grid.nodeLocal().putIfAbsent("lsnr", lsnr);

        // Make sure that we only subscribe one time regardless
        // of how many times we run the example.
        if (prev == null)
            grid.addLocalEventListener(lsnr,
                EVT_CACHE_OBJECT_PUT, 
                EVT_CACHE_OBJECT_READ, 
                EVT_CACHE_OBJECT_REMOVED);
     }
});

Note how easy it is in GridGain to execute any kind of code on all grid nodes (or any subset of nodes) without actually having to deploy anything. Now lets play with some basic cache operations and see what happens:

// Create strongly typed cache projection to avoid casting.
final GridCacheProjection<Integer, String> cache = 
    G.grid().cache().projection(Integer.class, String.class);

// Store some values in cache.
for (int i = 0; i < 10; i++)
    cache.put(i, "value-" + i);

// Note that size may differ depending on whether cache
// is distributed or partitioned.
System.out.println("Cache size: " + cache.size());

// Visit every cache element stored on local node.
// Note that 'CI1' is a just a type alias for 'GridInClosure' type.
cache.forEach(new CI1<GridCacheEntry<Integer, String>>() {
    @Override public void apply(GridCacheEntry<Integer, String> e) {
        // Peek at locally cached values.
        System.out.println("Visited locally cached entry: " + e.peek());
    }
});

// Collocate computations and data.
for (int i = 0; i < 10; i++) {
    final int key = i;

    // Find primary node for a key.
    final UUID nodeId = cache.mapKeyToNode(key);

    // Execute your computations on nodes where the data is cached to avoid a 
    // potentially heavy operation of bringing data to the local node. 
    // This is called Collocation of Computations and Data.
    G.grid().node(nodeId).run(UNICAST, new Runnable() {
        @Override public void run() {
            System.out.println("Collocating computations and data " + 
                "on node: " + nodeId);

            // Usually you would do something more complex than this :)
            System.out.println("Cached value: " + cache.peek(key));
        }
    });
}

// The 'get' operation will bring values from remote nodes
// even if they are not cached on local node. Generally,
// you would want to avoid it, if possible, as it may 
// create unnecessary data traffic.
for (int i = 0; i < 10; i++)
    System.out.println("Cached value: " + cache.get(i));

The example above is just a small sample of what you can do with GridGain data grid. Note that if the cache is configured to be replicated (which is default), then data will be replicated to all nodes and every node will get the same copy. If cache is partitioned, then only a designated primary node (and also backup nodes, if any) will get to cache a specific key-value pair.

Also note how easily we brought our computations to the nodes where the data is cached, as opposed to bringing the data to the computations. Performing computations without any unnecessary data movement (a.k.a. data noise) is one of the most important elements in achieving better scalability.

To run this example, startup a few stand alone GridGain nodes by executing GRIDGAIN_HOME/bin/ggstart.{sh|bat} script and watch what happens.

Enjoy!