通过Java提交多个hadoop作业

Question

我需要向Hadoop提交几个作业，这些作业都是相关的(这就是为什么它们由相同的驱动程序类启动)，但彼此完全独立。现在我是这样开始工作的:

int res = ToolRunner.run(new Configuration(), new MapReduceClass(params), args);

，它运行一个作业，获取返回代码，然后继续。

我想做的是提交几个这样的作业并行运行，检索每个作业的返回代码。

(对我来说)最明显的想法是启动几个线程，每个线程负责一个hadoop任务，但我想知道hadoop是否有更好的方法来完成这个任务?我没有任何编写并发代码的经验，所以我宁愿不花很多时间学习它的复杂性，除非这里有必要。

Answer 1

这可能是一个建议，但意味着代码，所以我将把它作为一个答案。

在这段代码(个人代码)中，我只是迭代一些变量，并多次提交一个作业(同一作业)。

使用job.waitForCompletion(false)将帮助您提交多个作业。

while (processedInputPaths < inputPaths.length) {
    if (processedInputPaths + inputPathsLimit < inputPaths.length) {
        end = processedInputPaths + inputPathsLimit - 1;
    } else {
        end = inputPaths.length - 1;
    }
    start = processedInputPaths;
    Job job = this.createJob(configuration, inputPaths, cycle, start, end, outputPath + "/" + cycle);
    boolean success = job.waitForCompletion(true);
    if (success) {
        cycle++;
        processedInputPaths = end + 1;
    } else {
        LOG.info("Cycle did not end successfully :" + cycle);
        return -1;
    }
}

Answer 2

psabbate的回答让我找到了我缺少的一些API。我是这样解决的:

在驱动类中，使用如下代码启动作业:

List<RunningJob> runningJobs = new ArrayList<RunningJob>();
for (String jobSpec: jobSpecs) {
    // Configure, for example, a params map that gets passed into the MR class's constructor
    ToolRunner.run(new Configuration(), new MapReduceClass(params, runningJobs), null);
}
for (RunningJob rj: runningJobs) {
    System.err.println("Waiting on job "+rj.getID());
    rj.waitForCompletion();
}

然后，在MapReduceClass中定义一个私有变量List<RunningJob> runningJobs，定义一个这样的构造函数:

public MergeAndScore(Map<String, String> p, List<RunningJob> rj) throws IOException {
    params = Collections.unmodifiableMap(p);
    runningJobs = rj;
}

在ToolRunner调用的run()方法中，定义您的JobConf并使用

提交作业。

JobClient jc = new JobClient();
jc.init(conf);
jc.setConf(conf);
runningJobs.add(jc.submitJob(conf));

这样，run()立即返回，并且可以通过驱动类中的runningJobs对象访问作业。

请注意，我正在使用旧版本的Hadoop，因此jc.init(conf)和/或jc.setConf(conf)可能是必要的，也可能不是必要的，这取决于您的设置，尽管可能至少需要其中一个。