为什么我的序列文件在我的hadoop映射器类中被读取两次？ [英] why is my sequence file being read twice in my hadoop mapper class?
问题描述
我有一个1264条记录的SequenceFile。每个密钥对于每条记录都是唯一的。我的问题是我的映射器似乎正在读这个文件两次，或者它正被读两次。为了理智检查，我已经写了一个小实用程序类来读取SequenceFile，事实上，只有1264条记录（即SequenceFile.Reader）。
i have a SequenceFile with 1264 records. each key is unique for each record. my problem is that my mapper seems to be reading this file twice or it is being read twice. for sanity checking, i have written a little utility class to read the SequenceFile and indeed, there are only 1264 records (i.e. SequenceFile.Reader).
每个Iterable只能得到1条记录，但是，当我迭代迭代器（Iterator）时，每个Key获得2条记录（每个键总是2个，而不是每个键1或3或其他东西）。
in my reducer, i should only get 1 record per Iterable, however, when i iterate over the iterable (Iterator), i get 2 records per Key (always 2 per key, and not 1 or 3 or something else per Key).
我的工作记录输出如下。我不确定为什么，但为什么处理的总输入路径是2？当我运行我的作业时，我试过了-Dmapred.input.dir = / data和-Dmapred.input.dir = / data / part-r-00000，但是仍然需要处理的总路径是2。
the logging output of my Job is below. i am not sure why, but why is it that the "Total input paths to process" is 2? when i run my Job, i tried -Dmapred.input.dir=/data and also -Dmapred.input.dir=/data/part-r-00000, but still, the total paths to process is 2.
任何想法都是值得赞赏的。
any ideas is appreciated.
12/03/01 05:28:30 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
12/03/01 05:28:30 INFO input.FileInputFormat: Total input paths to process : 2
12/03/01 05:28:31 INFO mapred.JobClient: Running job: job_local_0001
12/03/01 05:28:31 INFO input.FileInputFormat: Total input paths to process : 2
12/03/01 05:28:31 INFO mapred.MapTask: io.sort.mb = 100
12/03/01 05:28:31 INFO mapred.MapTask: data buffer = 79691776/99614720
12/03/01 05:28:31 INFO mapred.MapTask: record buffer = 262144/327680
12/03/01 05:28:31 INFO mapred.MapTask: Starting flush of map output
12/03/01 05:28:31 INFO mapred.MapTask: Finished spill 0
12/03/01 05:28:31 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
12/03/01 05:28:31 INFO mapred.LocalJobRunner:
12/03/01 05:28:31 INFO mapred.TaskRunner: Task 'attempt_local_0001_m_000000_0' done.
12/03/01 05:28:31 INFO mapred.MapTask: io.sort.mb = 100
12/03/01 05:28:31 INFO mapred.MapTask: data buffer = 79691776/99614720
12/03/01 05:28:31 INFO mapred.MapTask: record buffer = 262144/327680
12/03/01 05:28:31 INFO mapred.MapTask: Starting flush of map output
12/03/01 05:28:31 INFO mapred.MapTask: Finished spill 0
12/03/01 05:28:31 INFO mapred.TaskRunner: Task:attempt_local_0001_m_000001_0 is done. And is in the process of commiting
12/03/01 05:28:31 INFO mapred.LocalJobRunner:
12/03/01 05:28:31 INFO mapred.TaskRunner: Task 'attempt_local_0001_m_000001_0' done.
12/03/01 05:28:31 INFO mapred.LocalJobRunner:
12/03/01 05:28:31 INFO mapred.Merger: Merging 2 sorted segments
12/03/01 05:28:31 INFO mapred.Merger: Down to the last merge-pass, with 2 segments left of total size: 307310 bytes
12/03/01 05:28:31 INFO mapred.LocalJobRunner:
12/03/01 05:28:32 INFO mapred.TaskRunner: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting
12/03/01 05:28:32 INFO mapred.LocalJobRunner:
12/03/01 05:28:32 INFO mapred.TaskRunner: Task attempt_local_0001_r_000000_0 is allowed to commit now
12/03/01 05:28:32 INFO mapred.JobClient: map 100% reduce 0%
12/03/01 05:28:32 INFO output.FileOutputCommitter: Saved output of task 'attempt_local_0001_r_000000_0' to results
12/03/01 05:28:32 INFO mapred.LocalJobRunner: reduce > reduce
12/03/01 05:28:32 INFO mapred.TaskRunner: Task 'attempt_local_0001_r_000000_0' done.
12/03/01 05:28:33 INFO mapred.JobClient: map 100% reduce 100%
12/03/01 05:28:33 INFO mapred.JobClient: Job complete: job_local_0001
12/03/01 05:28:33 INFO mapred.JobClient: Counters: 12
12/03/01 05:28:33 INFO mapred.JobClient: FileSystemCounters
12/03/01 05:28:33 INFO mapred.JobClient: FILE_BYTES_READ=1320214
12/03/01 05:28:33 INFO mapred.JobClient: FILE_BYTES_WRITTEN=1275041
12/03/01 05:28:33 INFO mapred.JobClient: Map-Reduce Framework
12/03/01 05:28:33 INFO mapred.JobClient: Reduce input groups=1264
12/03/01 05:28:33 INFO mapred.JobClient: Combine output records=0
12/03/01 05:28:33 INFO mapred.JobClient: Map input records=2528
12/03/01 05:28:33 INFO mapred.JobClient: Reduce shuffle bytes=0
12/03/01 05:28:33 INFO mapred.JobClient: Reduce output records=2528
12/03/01 05:28:33 INFO mapred.JobClient: Spilled Records=5056
12/03/01 05:28:33 INFO mapred.JobClient: Map output bytes=301472
12/03/01 05:28:33 INFO mapred.JobClient: Combine input records=0
12/03/01 05:28:33 INFO mapred.JobClient: Map output records=2528
12/03/01 05:28:33 INFO mapred.JobClient: Reduce input records=2528
我的mapper类非常简单。它读入一个文本文件。
My mapper class is very simple. It reads in a text file. To each line, it appends "m" to the line.
public class MyMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
private static final Log _log = LogFactory.getLog(MyMapper.class);
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String s = (new StringBuilder()).append(value.toString()).append("m").toString();
context.write(key, new Text(s));
_log.debug(key.toString() + " => " + s);
}
}
我的reducer类也非常简单。它只是将r附加到行中。
My reducer class is also very simple. It simply appends "r" to the line.
public class MyReducer extends Reducer<LongWritable, Text, LongWritable, Text> {
private static final Log _log = LogFactory.getLog(MyReducer.class);
@Override
public void reduce(LongWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
for(Iterator<Text> it = values.iterator(); it.hasNext();) {
Text txt = it.next();
String s = (new StringBuilder()).append(txt.toString()).append("r").toString();
context.write(key, new Text(s));
_log.debug(key.toString() + " => " + s);
}
}
}
我的Job类如下
my Job class is as follows.
public class MyJob extends Configured implements Tool {
public static void main(String[] args) throws Exception {
ToolRunner.run(new Configuration(), new MyJob(), args);
}
@Override
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Path input = new Path(conf.get("mapred.input.dir"));
Path output = new Path(conf.get("mapred.output.dir"));
System.out.println("input = " + input);
System.out.println("output = " + output);
Job job = new Job(conf, "dummy job");
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
FileInputFormat.addInputPath(job, input);
FileOutputFormat.setOutputPath(job, output);
job.setJarByClass(MyJob.class);
return job.waitForCompletion(true) ? 0 : 1;
}
}
我的输入数据如下所示。
my input data looks like the following.
T, T
T, T
T, T
F, F
F, F
F, F
F, F
T, F
F, T
在运行我的Job之后，我得到如下输出。
after running my Job, i get an output like the following.
0 T, Tmr
0 T, Tmr
6 T, Tmr
6 T, Tmr
12 T, Tmr
12 T, Tmr
18 F, Fmr
18 F, Fmr
24 F, Fmr
24 F, Fmr
30 F, Fmr
30 F, Fmr
36 F, Fmr
36 F, Fmr
42 T, Fmr
42 T, Fmr
48 F, Tmr
48 F, Tmr
我是否在设置我的工作时出错？我尝试了以下方式来运行我的作业，并在这种方法中，文件只读取一次。为什么是这样？ System.out.println（inpath）和System.out.println（outpath）值是相同的！帮助？
did i do something wrong with setting up my Job? i tried the following way to run my Job, and in this approach, the file only gets read once. why is this? the System.out.println(inpath) and System.out.println(outpath) values are identical! help?
public class MyJob2 {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 2) {
System.err.println("Usage: MyJob2 <in> <out>");
System.exit(2);
}
String sInput = args[0];
String sOutput = args[1];
Path input = new Path(sInput);
Path output = new Path(sOutput);
System.out.println("input = " + input);
System.out.println("output = " + output);
Job job = new Job(conf, "dummy job");
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(MyMapper.class);
job.setReducerClass(MyReducer.class);
FileInputFormat.addInputPath(job, input);
FileOutputFormat.setOutputPath(job, output);
job.setJarByClass(MyJob2.class);
int result = job.waitForCompletion(true) ? 0 : 1;
System.exit(result);
}
}
推荐答案
我从hadoop邮件列表中获得帮助。我的问题是与下面的行。
i got help from the hadoop mailing list. my problem was with the line below.
FileInputFormat.addInputPath(job, input);
这行简单地将输入附加到配置中。在注释完此行后，输入文件现在只能读取一次。实际上，我也在其他行注释掉了，
this line simply appends input back to config. after commenting this line out, the input file is read only once now. in fact, i also commented out the other line,
FileOutputFormat.setOutputPath(job, output);
一切仍然有效。
这篇关于为什么我的序列文件在我的hadoop映射器类中被读取两次？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
