GridGain MapReduce clarity needed in function G.grid().reduce()

Question

I have used this example which is present in most of the sites..

int letterCnt = G.grid().reduce(
GridClosureCallMode.SPREAD,
F.yield("Counting Letters In This Phrase".split(" "),
    new C1<String, Integer>() {
        @Override public Integer apply(String word) {
            return word.length();
        }
    }
),
F.sumIntReducer()
);

The First argument denotes the equal distribution of work load( i assume its more like round robin basis) The Second argument has the code that will be executed in all the discovered nodes The Third which receives all result data from apply() executed in different nodes and processes accordingly.

I would like to know whether it is possible to substitute the third argument F.sumIntReducer() with a function of our own. If yes, I would like to see an example. Let's say creating the same function that has the same functionality of F.sumIntReducer() (i.e. sum's up all the length found by different nodes).

Answer 1

Yes, you can define your own custom reducer. The example is here.

Answer 2

The reduce method has global input and output data types. Its schematic structure is as follows:

1   resultValue = grid.reduce(
2   job distribution mode,
3   executable logic function,
4   split function,
5   collect function);

1. The result value of the task is assigned directly to the return value of the reduce function. Grid is the corresponding grid, where the job is sent to. Multiple grids may be defined and exist in parallel here.
2. The job distribution setting is defined here. The parameter for job distribution offers three options: -GridClosureCallMode.BALANCE – balanced (round robin?) distribution of jobs -GridClosureCallMode.BROADCAST – all nodes process all jobs -GridClosureCallMode.SPREAD – all jobs are randomly distributed
3. The atomic logical function is defined here. This part is called a job; it is send to a node, processed and contains a part of the global result. It has defined input and output data types which are supported. GridGain also supports the distribution of all necessary libraries for this function out of the box. This means the master node is not limited to use libraries, which all nodes have locally available, because all necessary libraries are shipped with the job. This creates of course more data traffic.
4. The input data needs to be distributed to the nodes. Each function is supplied with one of the data sets from the split function. The breakdown of the data is stored in an array list with the corresponding data type. Only low-level data types are supported, due to the implementation results (according to GridGain , also high level data should transferable). To transfer more complex data like PDFs and images, an encapsulation into byte arrays has to be done.
5. The master node uses this function to receive the resulting parts and reassemble them into the final result.

simple example: (does not utilize grid very much as only memory operations and not cpu intensive, so dont expect improvements to single execution)

private static int countLettersReducer(String phrase) throws GridException {

        // final GridLogger log = grid.log();
        int letterCount = 0;

        // Execute Hello World task.
        try {

            @SuppressWarnings("unchecked")
            int letterCnt =

            grid.reduce(GridClosureCallMode.BALANCE,
            // input: string
            // output: integer
                    new GridClosure<String, Integer>() {

                        private static final long serialVersionUID = 1L;

                        // Create executable logic block for a job part
                        @Override
                        public Integer apply(String word) {
                            // Print out a given word, just so we can
                            // see which node is doing what.
                            // System.out.println(">>> Calculating for word: " + word);
                            // Return the length of a given word, i.e. number of
                            // letters.
                            return word.length();
                        }
                    },

                    // split tasks for single jobs according to this function
                    // split at linebreaks
                    Arrays.asList(phrase.split("\n")),

                    // Collection of words.
                    // Collect the results from each job of the nodes
                    //input and output is integer
                    new GridReducer<Integer, Integer>() {
                        /**
                         * 
                         */
                        private static final long serialVersionUID = 1L;
                        private int sum;

                        @Override
                        public boolean collect(Integer res) {
                            sum += res;

                            return true; // True means continue collecting until
                                            // last result.
                        }

                        // return the collected results
                        @Override
                        public Integer apply() {
                            return sum;
                        }
                    });

            letterCount = letterCnt;
        } catch (Exception e) {

        }
        return letterCount;
    }