如何在异常的hadoop mapreduce中将任务状态更改为失败

Question

我正在使用 hadoop mapreduce 来完成以下任务：

我的映射器将从 HDFS 读取 .xml 文件并将其传递给某个服务。我的setup() 中有这样的代码：

try
{
    System.out.println(propertyName);
    session = FindPath.createSession("localhost",3250, EncodingConstants.en_ISO_8859_1);
    session.open();
}
catch 
{
    System.out.println("error");
}

当datanode没有FindPath [ FindPath - service is not yet started ] 时会抛出异常。

现在我的问题是我的映射程序没有移动到另一个正在运行 FindPath 服务的集群。

示例：

Datanode 1 Datanode 2 - 两个数据节点

如果 DataNode 1 尚未启动 FindPath 服务，则 Input 应移动到 Datanode 2。

我怎样才能做到这一点？

当Datanode 1 中的TaskStatus 抛出异常时，如何将其更改为“失败”？

更新

Job j;
catch(Exception Ex)
{   
    j.failTask((TaskAttemptID)context.getTaskAttemptID());
    System.out.println("error");
}

我使用过类似的东西，但它会引发 NullPointer 异常。

如何在我的映射器或映射器设置新 API 中使用failTask？

Answer 1

使用JobClient 访问RunningJob 类（我有1.0.4 API）。

所以代码看起来像这样：

在您的setup() 中有一个JobClient 和一个RunningJobreference。

方法如下：

public void setup(Context context)
{
    JobClient jobClient;
    RunningJob runningJob;

    try 
    {
        jobClient = new JobClient((JobConf)context.getConfiguration());
        runningJob = jobClient.getJob((JobID)(context.getJobId()); //mapred.JobID!
    }
    catch (IOException e)
    {
        System.out.println("IO Exception");
    }


    try
    {
        System.out.println(propertyName);
        session = FindPath.createSession("localhost",3250, EncodingConstants.en_ISO_8859_1);
        session.open();
    }
    catch 
    {
        System.out.println("error");
        runningJob.killTask((TaskAttemptID)context.getTaskAttemptID(), true);// cast as mapred.TaskAttemptID
    }

}

这会导致 TaskAttempt 失败。

最后，您可能应该将 mapred.map.max.attempts 设置为 1，这样失败的 taskAttempt 就是失败的任务。

注意：

您应该考虑更改 mapred.max.map.failures.percent，因为它反映了您的集群对失败任务的容忍度。