java - 实现Hadoop计数器时主线程中的IllegalStateException

Question

我已经在Hadoop中使用Counters来计算不同类型的LIC客户。下面是我的代码-

map

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;

public class Map 
            extends Mapper<LongWritable,Text,Text,IntWritable>
{

@Override
protected void map(LongWritable key,Text value,Context context) 
                        throws IOException,InterruptedException
{   
    String line = value.toString().trim();
    String[] part = line.split(" ");

    if(part.length > 1)
    {
        String str = part[0].toLowerCase();

        if(str.equals("LIC1"))
            context.getCounter("LIC1","lic1").increment(1);
        else if(str.equals("LIC2"))
            context.getCounter("LIC2", "lic2").increment(1);
        else
            context.getCounter("Others", "others").increment(1);

        context.write(new Text(str),new IntWritable(1));
    }
}
}

驾驶员等级

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class MapReduceDriver
{
public static void main(String[] args) throws Exception
{
    Job job = new Job();

    job.setJarByClass(MapReduceDriver.class);
    job.setJobName("Counter Example");

    FileInputFormat.addInputPath(job,new Path(args[0]));
    FileOutputFormat.setOutputPath(job,new Path(args[1]));

    job.setMapperClass(Map.class);
    job.setNumReduceTasks(0);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);

    long lic1 = job.getCounters().findCounter("LIC1", "lic1").getValue();
    long lic2 = job.getCounters().findCounter("LIC2", "lic2").getValue();
    long others = job.getCounters().findCounter("Others", 
         "others").getValue();

    System.out.println("LIC1= "+lic1);
    System.out.println("LIC2= "+lic2);
    System.out.println("Others= "+others);

    System.exit(job.waitForCompletion(true)?0:1);

}

}

驱动程序类抛出以下异常

Exception in thread "main" java.lang.IllegalStateException: Job in state 
DEFINE instead of RUNNING
at org.apache.hadoop.mapreduce.Job.ensureState(Job.java:64)
at org.apache.hadoop.mapreduce.Job.getCounters(Job.java:425)
at MapReduceDriver.main(MapReduceDriver.java:27)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke 
  (NativeMethodAccessorImpl.java:62)
  at 
  sun.reflect.DelegatingMethodAccessorImpl.invoke 
  (DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:498)
  at org.apache.hadoop.util.RunJar.main(RunJar.java:156

您能帮我找出我们的代码有什么问题吗？

Answer 1

job.getCounters()方法仅适用于正在运行的作业。而且我看不到提交作业以在MapReduceDriver类中运行的代码。添加job.submit()提交作业以运行到集群中并立即返回。