0

我知道这一定很简单，但我不知道如何获取 Hadoop JobTracker 的实例。在意识到我不能调用它的任何构造函数后，我试图在提交作业后像这样实例化它：

  JobClient client = new JobClient(conf);
  RunningJob runningJob = client.submitJob(conf);    
  JobTracker jobTracker = JobTracker.startTracker(_conf);

如果我把 jobTracker 线去掉，程序运行得很好。当我把它留在里面时，我得到一个异常，说 JobTracker 已经存在于那个 JobConf 中。

有谁知道我应该如何获得 JobTracker 的实例？我目前正在使用 hadoop 1.2.1。

编辑：我认为我得到的异常是因为 JobTracker 试图在已经有 JobTracker 监听的端口上启动。我可能会在不同的端口上启动 JobTracker 以进行测试运行，但是我的 Hadoop 集群将有 2 个 JobTracker，这不是它应该运行的方式。我很想在启动自己的 JobTracker 之前尝试停止现有的 JobTracker（这完全是在测试环境中，如果系统与其他 Jobs 共享，这将是不可行的），但这似乎走错了路。

4

1 回答 1

1

您无需在客户端代码中创建 JobTracker 的实例，您已经创建的 JobClient 具有可用于与正在运行的 JobTracker 交互的所有方法（这是通过调用 submitJob.

编辑

不幸的是，无法通过客户端 API 获取成功的任务主机名 - 令人沮丧的是，您可以通过命令行执行此操作，但它使用私有 API 调用，因此如果您要调用此命令和然后解析标准输出（您还将获得任务设置和清理事件）：

user@host1:~$ /opt/hadoop/default/bin/hadoop job -events job_201311110747_0001 0 100
Task completion events for job_201311110747_0001
Number of events (from 0) are: 4
SUCCEEDED attempt_201311110747_0001_m_000002_0 http://host1:50060/tasklog?plaintext=true&attemptid=attempt_201311110747_0001_m_000002_0
SUCCEEDED attempt_201311110747_0001_m_000000_0 http://host2:50060/tasklog?plaintext=true&attemptid=attempt_201311110747_0001_m_000000_0
SUCCEEDED attempt_201311110747_0001_r_000000_0 http://host3:50060/tasklog?plaintext=true&attemptid=attempt_201311110747_0001_r_000000_0
SUCCEEDED attempt_201311110747_0001_m_000001_0 http://host4:50060/tasklog?plaintext=true&attemptid=attempt_201311110747_0001_m_000001_0

一个选项可能是使用一些基于反射的黑客技术使私有 API 公开可见，然后根据需要使用 - 作为参考，这里是您需要在代码中复制上述内容的 API 调用（这可能与不同版本不向前或向后兼容Hadoop - 这里是 1.2.1):

public class JobClientDriver extends Configured implements Tool {
    public static void main(String args[]) throws Exception {
        ToolRunner.run(new JobClientDriver(), args);
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();

        JobClient client = new JobClient(new JobConf(conf));

        Method method = JobClient.class.getDeclaredMethod("createRPCProxy", InetSocketAddress.class,
                Configuration.class);
        method.setAccessible(true);

        Object rpcClientSubProtocol = method.invoke(client, JobTracker.getAddress(conf), conf);

        Method completeEventsMethod = rpcClientSubProtocol.getClass().getDeclaredMethod("getTaskCompletionEvents",
                JobID.class, int.class, int.class);

        for (Object tceObj : ((Object[]) completeEventsMethod.invoke(rpcClientSubProtocol,
            JobID.forName("job_201311110747_0001"), 0, 100))) {
            TaskCompletionEvent tce = (TaskCompletionEvent) tceObj;
            if (tce.isMapTask()) {
                URI uri = new URI(tce.getTaskTrackerHttp());
                System.err.println(tce.getTaskAttemptId() + " @ " + uri.getHost());
            }
        }

        return 0;
    }
}
于 2013-11-10T00:15:30.320 回答