В работе Map-Reduce я использую пять разных файлов, где в моем наборе данных содержатся значения в двух категориях P
и I
. После того, как I
определенные значения найдены, я передаю их в файл I-part-r-00000 и, соответственно, для P. Я использую класс MultipleOutputformat в редукторе для достижения этой цели.
Класс My Mapper содержит:
public class parserMapper extends Mapper<LongWritable, Text, Text, Text> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String IPFLAG = "";
String[] element_data= value.toString.split(",");
if(element_data[0].toString().trim().equalsIgnoreCase("005010X222A1")){
IPFLAG = "P";
}
else {
IPFLAG = "I";
}
if (IPFLAG == "P") {
context.write(new Text(IPFLAG), new Text(theData));
}
else if (IPFLAG == "I") {
context.write(new Text(IPFLAG), new Text(theData));
}
else{
System.out.println("No category found");
}
}
public void run(Context context) throws IOException, InterruptedException {
setup(context);
while (context.nextKeyValue()) {
map(context.getCurrentKey(), context.getCurrentValue(), context);
}
cleanup(context);
}
}
Класс редуктора включает в себя:
public class parserReducer extends Reducer<Text, Text, Text, Text> {
private MultipleOutputs multipleOutputs;
@Override
protected void setup(Context context) throws IOException, InterruptedException {
multipleOutputs = new MultipleOutputs(context);
}
@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
multipleOutputs.close();
}
@Override
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
Object c = null;
try{
if (!(key.toString().isEmpty())) {
for (Text value : values) {
multipleOutputs.write(c, value, key.toString());
}
}
}
catch(Exception e){ System.out.println("Caught Exception: " + e.getMessage());}
}
}
и код драйвера включает в себя =>
public class parserDriver {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("textinputformat.record.delimiter", "~"+"\n"+"ISA*");
Job job = new Job(conf);
job.setJobName("PARSER");
job.setJarByClass(parserDriver.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(parserMapper.class);
job.setReducerClass(parserReducer.class);
// job.setOutputFormatClass(TextOutputFormat.class);
LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
// job.setOutputFormatClass(LazyOutputFormat.class);
/* MultipleOutputs.addNamedOutput(job, "P", TextOutputFormat.class, Text.class, Text.class);
MultipleOutputs.addNamedOutput(job, "I", TextOutputFormat.class, Text.class, Text.class);
*/
// Pass as option -D mapred.reduce.tasks=<number>
job.setNumReduceTasks(3);
/* This line is to accept the input recursively */
//FileInputFormat.setInputDirRecursive(job, true);
FileInputFormat.addInputPath(job, "/Users/Mohit/input");
FileOutputFormat.setOutputPath(job, "/Users/Mohit/output");
/*
* Delete output file path if already exists
*/
FileSystem fs = FileSystem.get(conf);
if (fs.exists(outputFilePath)) {
fs.delete(outputFilePath, true);
}
return job.waitForCompletion(true) ? 0: 1;
}
}
Через все это я пытаюсь добиться двух разделов для одного файла
file1 -> P-part-r00000, I-part-r00001
file2 -> P-part-r00002, I-part-r00003
. но я получаю два раздела против всех файлов, которые вводятся в качестве входных данных для этой работы.
file1, file2, file3, file4, file5 -> P-part-r00000, I-part-r00001
Не уверен, что мне здесь не хватает, если кто-нибудь может помочь, пожалуйста?