Cách triển khai sắp xếp trong hadoop?

Question

Vấn đề của tôi là sắp xếp các giá trị trong một tệp. khóa và giá trị là số nguyên và cần duy trì khóa của giá trị được sắp xếp.

key value 
1  24 
3  4 
4  12 
5  23 

đầu ra:

1  24 
5  23 
4  12 
3  4 

tôi đang làm việc với dữ liệu khổng lồ và phải chạy mã trong một cụm máy hadoop. Làm cách nào để tôi có thể thực hiện điều đó với mapreduce?

Answer 1

Bạn có lẽ có thể làm được điều này (tôi giả sử bạn đang sử dụng Java ở đây)

Từ bản đồ phát ra như thế này -

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

Vì vậy, tất cả các bạn những giá trị mà cần phải được sắp xếp nên là trong công việc tạo bản đồ của bạn. Hadoop theo mặc định sắp xếp theo thứ tự tăng dần của khóa.

Do đó, hoặc là bạn làm điều này để sắp xếp theo thứ tự giảm dần,

job.setSortComparatorClass(LongWritable.DecreasingComparator.class); 

Hoặc, này,

Bạn cần phải thiết lập một phong tục giảm dần Sắp xếp sánh, mà đi một cái gì đó như thế này trong công việc của bạn.

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); 
    } 
} 

Các suffle và sắp xếp giai đoạn trong Hadoop sẽ chăm sóc sắp xếp phím của bạn để 24,4,12,23

giảm dần Sau khi bình luận:

Nếu bạn cần một dần IntWritable So sánh, bạn có thể tạo một và sử dụng nó như thế này -

job.setSortComparatorClass(DescendingIntComparable.class); 

Trong trường hợp nếu bạn đang sử dụng JobConf, sử dụng để thiết lập

jobConfObject.setOutputKeyComparatorClass(DescendingIntComparable.class); 

Đặt đoạn mã sau bên dưới chức năng main() của bạn -

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); 
     } 
    } 
}