Цель состоит в том, чтобы узнать, насколько файл X похож на файлы y1, y2, ..., yn.
Для каждого файла я извлекаю информацию и храню ее в структурах;скажем, из файла я делаю подсчет слов и сохраняю результаты в HashMap<String, Integer> wordCount (есть другие структуры, хранящие другую информацию).
Так что мне нужно сгенерировать wordCount из fileX;извлекать wordCount из файла Y (предварительно сгенерированного и записанного в файлы HDFS);посчитайте, насколько эти два слова похожи (я не могу сделать построчное различие; мне нужно сходство в процентах).
FileX фиксирован и его необходимо сравнить с N fileY.
Итак, моя идея заключалась в следующем:
Job1: вычислить информацию о fileX и записать ее в HDFS.
Job2 (chainMapper map1-map2):
Map1: прочитать HashMap<String, Integer> wordCount из файлаX;передача структур в Map2.
Map2: получает 2 входа, структуры fileX, путь к каталогу fileYs.
Map2 вычисляет сходство HashMap<String, Integer> wordCountX и HashMap<String, Integer> wordCountY;редуктор получает все значения сходства и упорядочивает их.
Я читал на Hadoop - The definitive guide of Tom White и в Интернете тоже о MultipleInputs, но речь идет не о двух входах в 1 маппер, а о дифференцировании мапперов на основевходы.Поэтому я хочу спросить, как переслать два значения в один картограф;Я рассмотрел использование распределенного кэша, но это не что-то полезное для этой проблемы;и наконец, как быть уверенным, что каждый маппер получает свой файл Y.
Я пытался обновить глобальный HashMap<String, Integer> wordCount, но когда начинается новое задание, он не может получить доступ к этой структуре (или, лучше, он пуст).
public class Matching extends Configured implements Tool{
private static HashMap<String, Integer> wordCountX;
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Matching(), args);
System.exit(res);
} //end main class
public int run(String[] args) throws Exception {
...
}
}
РЕДАКТИРОВАТЬ:
Ответ вот это хорошее решение.
Я добавляю результирующий фрагмент кода.
Запуск задания:
//configuration and launch of job
Job search = Job.getInstance(getConf(), "2. Merging and searching");
search.setJarByClass(this.getClass());
MultipleInputs.addInputPath(search, creationPath, TextInputFormat.class);
MultipleInputs.addInputPath(search, toMatchPath, TextInputFormat.class);
FileOutputFormat.setOutputPath(search, resultPath);
search.setNumReduceTasks(Integer.parseInt(args[2]));
search.setMapperClass(Map.class);
search.setReducerClass(Reduce.class);
search.setMapOutputKeyClass(ValuesGenerated.class);
search.setMapOutputValueClass(IntWritable.class);
//TODO
search.setOutputKeyClass(NullWritable.class);
search.setOutputValueClass(Text.class);
return search.waitForCompletion(true) ? 0 : 1;
Слияние карт (в фазе очистки):
@Override
public void cleanup(Context context) throws IOException, InterruptedException {
InputSplit split = context.getInputSplit();
Class<? extends InputSplit> splitClass = split.getClass();
FileSplit fileSplit = null;
if (splitClass.equals(FileSplit.class)) {
fileSplit = (FileSplit) split;
} else if (splitClass.getName().equals(
"org.apache.hadoop.mapreduce.lib.input.TaggedInputSplit")) {
// begin reflection hackery...
try {
Method getInputSplitMethod = splitClass
.getDeclaredMethod("getInputSplit");
getInputSplitMethod.setAccessible(true);
fileSplit = (FileSplit) getInputSplitMethod.invoke(split);
} catch (Exception e) {
// wrap and re-throw error
throw new IOException(e);
}
// end reflection hackery
}
String filename = fileSplit.getPath().getName();
boolean isKnown;
/*
the two input files are nominated dinamically;
the file0 has some name "023901.txt",
the file1 is the output of a precedent MR job, and is
something like "chars-r-000000"
*/
if(filename.contains(".txt")) {
isKnown = false;
}
else {
isKnown = true;
}
if(isKnown) { //file1, known
ValuesGenerated.setName(new Text(name));
//other values set
//...
context.write(ValuesGenerated, new IntWritable(1));
}
else { //file0, unknown
ValuesGenerated.setName(new Text("unknown"));
//other values set
//...
context.write(ValuesGenerated, new IntWritable(0));
}
}
Фаза сокращения:
public static class Reduce extends Reducer<ValuesGenerated, IntWritable, NullWritable, Text> {
@Override
public void reduce(ValuesGenerated key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
ValuesGenerated known;
ValuesGenerated unk;
String toEmit = null;
for (IntWritable value : values) {
if(value.get() == 1) { //known
known = key;
toEmit = key.toString();
toEmit += "\n " + value;
context.write(NullWritable.get(), new Text(toEmit));
}
else { //unknown
unk = key;
toEmit = key.toString();
toEmit += "\n " + value;
context.write(NullWritable.get(), new Text(toEmit));
}
}
}//end reduce
} //end Reduce class
Я столкнулся с другой проблемой, но обошел ее с этим решением Hadoop MultipleInputs терпит неудачу с ClassCastException