Как реализовать сортировку в hadoop?

Question

Моя проблема заключается в сортировке значений в файле. ключи и значения являются целыми числами и должны содержать ключи отсортированных значений.

key value 
1  24 
3  4 
4  12 
5  23 

выход:

1  24 
5  23 
4  12 
3  4 

Я работаю с массивными данными и должен запустить код в кластере Hadoop машин. Как я могу сделать это с помощью mapreduce?

Answer 1

Вы, вероятно, может сделать это (я предполагаю, что вы используете Java здесь)

С карты излучали как это -

context.write(24,1); 
context.write(4,3); 
context.write(12,4) 
context.write(23,5) 

Таким образом, все, что вам значения, которые должны быть отсортированы должны быть введите ключевое слово в вашей задаче mapreduce. Hadoop по умолчанию сортирует по возрастанию.

Следовательно, либо вы делаете это, чтобы отсортировать в порядке убывания,

job.setSortComparatorClass(LongWritable.DecreasingComparator.class); 

Или это,

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

public static class DescendingKeyComparator extends WritableComparator { 
    protected DescendingKeyComparator() { 
     super(Text.class, true); 
    } 

    @SuppressWarnings("rawtypes") 
    @Override 
    public int compare(WritableComparable w1, WritableComparable w2) { 
     LongWritable key1 = (LongWritable) w1; 
     LongWritable key2 = (LongWritable) w2;   
     return -1 * key1.compareTo(key2); 
    } 
} 

Суффль и родом этап в Hadoop будут заботиться о сортировке ключей в порядке 24,4,12,23

убыванию После комментария:

Если вам требуется нисходящая IntWritable Сопоставимые, вы можете создать и использовать его как это -

job.setSortComparatorClass(DescendingIntComparable.class); 

В случае, если вы используете JobConf, использовать это, чтобы установить

jobConfObject.setOutputKeyComparatorClass(DescendingIntComparable.class); 

Поместите следующий код ниже вашей main() функции -

public static void main(String[] args) { 
    int exitCode = ToolRunner.run(new YourDriver(), args); 
    System.exit(exitCode); 
} 

//this class is defined outside of main not inside 
public static class DescendingIntWritableComparable extends IntWritable { 
    /** A decreasing Comparator optimized for IntWritable. */ 
    public static class DecreasingComparator extends Comparator { 
     public int compare(WritableComparable a, WritableComparable b) { 
      return -super.compare(a, b); 
     } 
     public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) { 
      return -super.compare(b1, s1, l1, b2, s2, l2); 
     } 
    } 
}