Как сравнить информацию из разных файлов в MapReduce?

Цель состоит в том, чтобы узнать, насколько файл X похож на файлы y1, y2, ..., yn.

Для каждого файла я извлекаю информацию и сохраняю ее в структурах; скажем, из файла я подсчитываю количество слов и сохраняю результаты в HashMap<String, Integer> wordCount (есть другие структуры, хранящие другую информацию).

Поэтому мне нужно сгенерировать wordCount из fileX; извлечь wordCount из файла Y (предварительно сгенерированного и записанного в файлы HDFS); подсчитайте, насколько эти два количества слов похожи (я не могу сделать построчную разницу, мне нужно сходство в процентах).

FileX исправлен, и его необходимо сравнить с N файломY.

Итак, моя идея была:

Job1: вычислить информацию fileX и записать ее в HDFS.

Job2 (chainMapper карты1-карты2):

Map1: прочитать HashMap<String, Integer> wordCount файла X; передача структур в Map2.

Map2: получает 2 входа, структуры файла X, путь к каталогу файлов Y.

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