Come implementare l'ordinamento in hadoop?

Question

Il mio problema è l'ordinamento dei valori in un file. Le chiavi e i valori sono numeri interi e devono mantenere le chiavi dei valori ordinati.

key value 
1  24 
3  4 
4  12 
5  23 

uscita:

1  24 
5  23 
4  12 
3  4 

sto lavorando con dati di massa e deve eseguire il codice in un cluster di macchine Hadoop. Come posso farlo con mapreduce?

Answer 1

Probabilmente si può fare questo (sto supponendo che si sta utilizzando Java qui)

Da mappe emettono come questo -

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

Quindi, tutti i valori che devono essere ordinati dovrebbe essere il inserisci il tuo lavoro mapreduce. Hadoop di default ordina in ordine crescente di chiave.

Quindi, o si fa questo per ordinare in ordine decrescente,

job.setSortComparatorClass(LongWritable.DecreasingComparator.class); 

Oppure, questo,

è necessario impostare una consuetudine Descending Ordina comparatore, che va qualcosa di simile nel tuo lavoro.

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

Il suffle e una sorta di fase in Hadoop si prenderà cura di ordinare le chiavi in ​​ordine 24,4,12,23

decrescente Dopo commento:

Se è necessario un Discendente IntWritable paragonabile, è possibile crearne uno e utilizzarlo come questo -

job.setSortComparatorClass(DescendingIntComparable.class); 

Nel caso in cui se si utilizza JobConf, utilizzare questo per impostare

jobConfObject.setOutputKeyComparatorClass(DescendingIntComparable.class); 

inserire il seguente codice sotto la vostra funzione 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); 
     } 
    } 
}