Наконец то я воплотил давнюю идею написать универсальную программку которая позволяет распараллеливать обработку файлов на GridGain.Идея очень простая - универсальная task и job которые выполняют данную команду с некоторыми параметрами на разных машинах и собирает output с этих команд а потом сборка результата с помощью простой команды.Итак чуть более формально:Задача: посчитать количество вхождений строки во все файлы в определенном каталоге на сетевом диске (задача распараллеливания становится очень актуальна если скажем у вас 1000 файлов по 1 гигабайту например). (Кстати не меняя Java кода вы можете выполнять любые скрипты: shell, cmd, perl что угодно)
Как это работает:Вы стартуете универсальный лоадер GridCmdLineLoader как обычный класс с методом main() и передаете ему 3 параметра:
a) Файл конфигурацию грида
б) Команду, которая будет обрабатывать файлы и выводить в консоль результат работы
с) Команду, которая из результатов пункта б) сделает окончательный результат.
Пример запуска:java GridCmdLineLoader config/default-spring.xml "search.sh /mount/netdrive/data searchString" result.shПрошу заметить что это просто пример :) конечно же надо указать все пути правильно, указать classpath Java машине и прочее.Теперь код:1) Создадим скрипт (для примера на bash) который умеет искать подстроку в файлах, но немного хитрее - у него будет возможность "распараллеливания". То есть проще говоря он будет обрабатывать не все файлы, а только некоторые на основе 2 переменных окружения: GRID_JOB_NUM и GRID_TOTAL_JOBS. Прошу заметить что скрипт выводит результат с помощью echo. Это и есть результат его работы. Скрипт назовем search.sh и пусть он принимает 2 параметра. 1-й это имя каталога а 2-й это строка для поиска.search.sh
#!/bin/bash
# First script parameter is search dir
searchDir=$1
# Second is a word.
searchWord=$2
result=0;
# Variables predefined by task
totalJobs=$GRID_TOTAL_JOBS
jobNum=$GRID_JOB_NUM
# Initialize then with default values if script started manually.
if [[ -z $totalJobs ]]; then
totalJobs=1
fi
if [[ -z $jobNum ]]; then
jobNum=0
fi
# Get list of files.
files=`ls $searchDir`
i=0
# One by one count number of given word using grep.
for file in $files; do
# Next file number that should be processed. Note that if
# number of jobs is 2 for example then every job will
# process every second file with $jobNum shift.
nextNum=$((i / totalJobs * totalJobs + jobNum))
if [[ $i = $nextNum ]]; then
cnt=`grep -c $searchWord $searchDir/$file`
result=$((result+cnt))
fi
i=$((i+1))
done
echo -n $result
exit 0;
2) Создаем еще простой скрипт опять-таки на bash. Его задача "собрать" вместе результаты выполнения первого скрипта на разных машинах (просто просуммировать их). Назовем его result.sh и его параметрами будут результаты выполнения search.sh на разных машинах. result.sh#!/bin/bash
result=0
until [ -z "$1" ]
do
result=$((result+$1))
shift
done
echo -n $result
3) Теперь самое интересное - Это так называемый loader. Несмотря на то что GridGain поставляется с различными loaders (для WebLogic, GlassFish, WebSphere, Servlet, Command line) нам пригодится более специализированный. А по сути это обычный класс с методом main(), который поднимает грид, выполняет нашу задачу и останавливает грид :)
Ниже идет класс названный GridCmdLineLoader и реализующий эту логику.
GridCmdLineLoader.javapackage org.gridgain.examples.cmdline;
import org.gridgain.grid.*;
import org.gridgain.grid.loaders.*;
@SuppressWarnings({"CallToSystemExit"})
@GridLoader(description = "Command line loader")
public final class GridCmdLineLoader {
/**
* Enforces singleton.
*/
private GridCmdLineLoader() {
// No-op.
}
/**
* Main entry point.
*
* @param args Command line arguments.
*/
@SuppressWarnings({"unchecked"})
public static void main(String[] args) {
if (args.length < 3) {
System.out.println("Too few arguments. Expected at least" +
" grid configuration target script and result script.");
System.exit(-1);
}
Grid grid = null;
Exception ex = null;
try {
grid = GridFactory.start(args[0]);
GridCmdLineTask task = new GridCmdLineTask(args[1], args[2]);
String res = grid.execute(task, null).get();
System.out.println(res);
}
catch (GridException e) {
ex = e;
}
finally {
if (grid != null) {
GridFactory.stop(grid.getName(), false);
}
}
if (ex != null) {
System.out.println("Got exception: " + ex);
System.exit(-1);
}
else {
System.exit(0);
}
}
}
4) Ну и последнее - grid task/job и пожалуй самое интересное - прошу заметитть что эта job умеет выполнять произвольный скрипт и все что скрипт выводит на консоль берет как строку и считает результатом работы - так что при желании вы можете не только считать но и возвращать результат поиска и все что хотите :)GridCmdLineTask.java
package org.gridgain.examples.cmdline;
import java.io.*;
import java.util.*;
import org.gridgain.grid.*;
import org.gridgain.grid.logger.*;
import org.gridgain.grid.resources.*;
public class GridCmdLineTask extends GridTaskSplitAdapter<String, String> {
/** */
public static final String GRID_JOB_NUM = "GRID_JOB_NUM";
/** */
public static final String GRID_TOTAL_JOBS = "GRID_TOTAL_JOBS";
/** */
@GridLoggerResource
private GridLogger log = null;
/** Executable command. */
private final String cmd;
/** Result reducing command. */
private final String resCmd;
/**
* @param cmd Command to execute.
* @param resCmd Command that will reduce results
*/
public GridCmdLineTask(String cmd, String resCmd) {
super();
this.cmd = cmd;
this.resCmd = resCmd;
}
/**
* {@inheritDoc}
*/
@Override
public Collection<? extends GridJob> split(int gridSize, String arg) {
Collection<GridJob> refs = new ArrayList<GridJob>(gridSize);
for (int i = 0; i < job =" new"> results) throws GridException {
String params = "";
for (GridJobResult res : results) {
params = params + res.getData() + ' ';
}
if (log.isDebugEnabled() == true) {
log.debug("Reducing results. Execute command '"
+ resCmd + ' ' + params + '\'');
}
// Make a call of reducing command.
return executeCmd(resCmd + ' ' + params, null, log);
}
/**
*
* @param params command parameters
* @param env command environment.
* @param log Grid logger.
* @return command output.
* @throws GridException If command execution failed.
*/
private static String executeCmd(String params, String[] env, GridLogger log)
throws GridException {
try {
Process proc = Runtime.getRuntime().exec(params, env);
proc.waitFor();
StringBuffer sb = new StringBuffer();
InputStream is = proc.getInputStream();
while (is.available() != 0) {
sb.append((char)is.read());
}
if (log.isDebugEnabled() == true) {
log.debug("Command execution result is " + sb.toString());
}
return sb.toString();
}
catch (IOException e) {
throw new GridException(e);
}
catch (InterruptedException e) {
throw new GridException(e);
}
}
/**
*
*/
public static class GridCmdLineJob extends GridJobAdapter<String> {
/** */
@GridLoggerResource
private GridLogger log = null;
/** Job number. */
private final long jobNum;
/** Number of jobs. */
private final long totalJobs;
/**
* Creates new command line job.
*
* @param jobNum Job number on a list of jobs.
* @param totalJobs Total jobs number.
*/
public GridCmdLineJob(long jobNum, long totalJobs) {
super();
this.jobNum = jobNum;
this.totalJobs = totalJobs;
}
/**
* {@inheritDoc}
*/
public Serializable execute() throws GridException {
String[] env = new String[]{GRID_JOB_NUM + "=" + jobNum,
GRID_TOTAL_JOBS + "=" + totalJobs};
if (log.isDebugEnabled() == true) {
log.debug("Executing given command line '" + getArgument() + '\'');
}
return executeCmd(getArgument(), env, this.log);
}
}
}
