Отсортированное количество слов с использованием Hadoop MapReduce

В простой программе сокращения количества слов на карте вывод, который мы получаем, сортируется по словам. Пример вывода может быть:
Яблоко 1
Мальчик 30
Кошка 2
Лягушка 20
Зебра 1
Если вы хотите, чтобы вывод был отсортирован по количеству вхождений слов, т.е. в формате ниже
1 Apple
1 Zebra
2 Cat
20 Frog
30 Boy
Вы можете создать другую программу MR, используя приведенный ниже преобразователь и редуктор, где ввод будет выводом получил от простой программы подсчета слов.

class Map1 extends MapReduceBase implements Mapper<Object, Text, IntWritable, Text>
{
    public void map(Object key, Text value, OutputCollector<IntWritable, Text> collector, Reporter arg3) throws IOException 
    {
        String line = value.toString();
        StringTokenizer stringTokenizer = new StringTokenizer(line);
        {
            int number = 999; 
            String word = "empty";

            if(stringTokenizer.hasMoreTokens())
            {
                String str0= stringTokenizer.nextToken();
                word = str0.trim();
            }

            if(stringTokenizer.hasMoreElements())
            {
                String str1 = stringTokenizer.nextToken();
                number = Integer.parseInt(str1.trim());
            }

            collector.collect(new IntWritable(number), new Text(word));
        }

    }

}


class Reduce1 extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text>
{
    public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable, Text> arg2, Reporter arg3) throws IOException
    {
        while((values.hasNext()))
        {
            arg2.collect(key, values.next());
        }

    }

}

Выходные данные примера количества слов Hadoop MapReduce сортируются по ключу. Таким образом, вывод должен быть в алфавитном порядке.

С помощью Hadoop вы можете создавать свои собственные ключевые объекты, которые реализуют интерфейс WritableComparable, позволяя переопределять метод compareTo. Это позволяет контролировать порядок сортировки.

Чтобы создать вывод, отсортированный по количеству вхождений, вам, вероятно, придется добавить еще одно задание MapReduce для обработки вывода из первого, как вы сказали. Эта вторая работа будет очень простой, возможно, даже не потребует фазы сокращения. Вам просто нужно реализовать свой собственный ключевой объект Writable, чтобы обернуть слово и его частоту. Пользовательская запись выглядит примерно так:

 public class MyWritableComparable implements WritableComparable {
       // Some data
       private int counter;
       private long timestamp;

       public void write(DataOutput out) throws IOException {
         out.writeInt(counter);
         out.writeLong(timestamp);
       }

       public void readFields(DataInput in) throws IOException {
         counter = in.readInt();
         timestamp = in.readLong();
       }

       public int compareTo(MyWritableComparable w) {
         int thisValue = this.value;
         int thatValue = ((IntWritable)o).value;
         return (thisValue < thatValue ? -1 : (thisValue==thatValue ? 0 : 1));
       }
     }

Я взял этот пример из здесь.

В Hadoop сортировка выполняется между фазами Map и Reduce. Одним из подходов к сортировке по вхождению слова может быть использование специального компаратора групп, который ничего не группирует; следовательно, каждый вызов сокращения — это просто ключ и одно значение.

public class Program {
   public static void main( String[] args) {

      conf.setOutputKeyClass( IntWritable.class);
      conf.setOutputValueClass( Text.clss);
      conf.setMapperClass( Map.class);
      conf.setReducerClass( IdentityReducer.class);
      conf.setOutputValueGroupingComparator( GroupComparator.class);   
      conf.setNumReduceTasks( 1);
      JobClient.runJob( conf);
   }
}

public class Map extends MapReduceBase implements Mapper<Text,IntWritable,IntWritable,Text> {

   public void map( Text key, IntWritable value, OutputCollector<IntWritable,Text>, Reporter reporter) {
       output.collect( value, key);
   }
}

public class GroupComaprator extends WritableComparator {
    protected GroupComparator() {
        super( IntWritable.class, true);
    }

    public int compare( WritableComparable w1, WritableComparable w2) {
        return -1;
    }
}

Как вы сказали, одна из возможностей - написать две работы для этого. Первая работа: простой пример подсчета слов

Вторая работа: выполняет сортировочную часть.

Псевдокод может быть:

Примечание. Выходной файл, созданный первым заданием, будет входным файлом для второго задания.

    Mapper2(String _key, Intwritable _value){
    //just reverse the position of _value and _key. This is useful because reducer will get the output in the sorted and shuffled manner.
    emit(_value,_key);
    }

    Reduce2(IntWritable valueofMapper2,Iterable<String> keysofMapper2){
//At the reducer side, all the keys that have the same count are merged together.
        for each K in keysofMapper2{
        emit(K,valueofMapper2); //This will sort in ascending order.
        }

    }

Вы также можете отсортировать в порядке убывания, для которого можно написать отдельный класс компаратора, который сделает свое дело. Включите компаратор внутри задания как:

Job.setComparatorclass(Comparator.class);

Этот компаратор будет сортировать значения в порядке убывания перед отправкой на сторону редуктора. Итак, на редюсере вы просто испускаете значения.