【发布时间】：2012-02-20 04:33:04
【问题描述】：
我的地图缩减结构
public class ChainingMapReduce {
public static class ChainingMapReduceMapper
extends Mapper<Object, Text, Text, IntWritable>{
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
// code
}
}
}
public static class ChainingMapReduceReducer
extends Reducer<Text,IntWritable,Text,IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {
//code
}
}
public static class ChainingMapReduceMapper1
extends Mapper<Object, Text, Text, IntWritable>{
public void map(Object key, Text value, Context context
) throws IOException, InterruptedException {
//code
}
}
}
public static class ChainingMapReduceReducer1
extends Reducer<Text,IntWritable,Text,IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values,
Context context
) throws IOException, InterruptedException {
//code
}
}
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
Configuration conf = new Configuration();
Job job = new Job(conf, "First");
job.setJarByClass(ChainingMapReduce.class);
job.setMapperClass(ChainingMapReduceMapper.class);
job.setCombinerClass(ChainingMapReduceReducer.class);
job.setReducerClass(ChainingMapReduceReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("/home/Desktop/log"));
FileOutputFormat.setOutputPath(job, new Path("/home/Desktop/temp/output"));
job.waitForCompletion( true );
System.out.println("First Job Completed.....Starting Second Job");
System.out.println(job.isSuccessful());
/* FileSystem hdfs = FileSystem.get(conf);
Path fromPath = new Path("/home/Desktop/temp/output/part-r-00000");
Path toPath = new Path("/home/Desktop/temp/output1");
hdfs.rename(fromPath, toPath);
conf.clear();
*/
if(job.isSuccessful()){
Configuration conf1 = new Configuration();
Job job1 = new Job(conf1,"Second");
job1.setJarByClass(ChainingMapReduce.class);
job1.setMapperClass(ChainingMapReduceMapper1.class);
job1.setCombinerClass(ChainingMapReduceReducer1.class);
job1.setReducerClass(ChainingMapReduceReducer1.class);
job1.setOutputKeyClass(Text.class);
job1.setOutputValueClass(IntWritable.class);
FileInputFormat.addInputPath(job, new Path("/home/Desktop/temp/output/part-r-00000)");
FileOutputFormat.setOutputPath(job, new Path("/home/Desktop/temp/output1"));
System.exit(job1.waitForCompletion(true) ? 0 : 1);
}
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
当我运行这个程序时......第一个作业得到完美执行，然后出现以下错误：
第一个工作完成.....开始第二个工作 true
12/01/27 15:24:21 INFO jvm.JvmMetrics：无法初始化 JVM 指标 使用 processName=JobTracker, sessionId= - 已经初始化 12/01/27 15:24:21 WARN mapred.JobClient：使用 GenericOptionsParser 进行解析 论据。应用程序应该实现同样的工具。 12/01/27 15:24:21 WARN mapred.JobClient：没有设置作业 jar 文件。用户 可能找不到类。请参阅 JobConf(Class) 或 JobConf#setJar（字符串）。 12/01/27 15:24:21 信息 mapred.JobClient: 清理暂存区 文件：/tmp/hadoop/mapred/staging/4991311720439552/.staging/job_local_0002 线程“主”中的异常 org.apache.hadoop.mapred.InvalidJobConfException：输出目录不是 放。在 org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.checkOutputSpecs(FileOutputFormat.java:123) 在 org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:872) 在 org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:833) 在 java.security.AccessController.doPrivileged(Native Method) 在 javax.security.auth.Subject.doAs(Subject.java:396) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1127) 在 org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:833) 在 org.apache.hadoop.mapreduce.Job.submit(Job.java:476) 在 org.apache.hadoop.mapreduce.Job.waitForCompletion(Job.java:506) 在 ChainingMapReduce.main(ChainingMapReduce.java:129)
我尝试对两个作业使用“conf”，对各自的作业使用“conf”“conf1”。
【问题讨论】：