Поиск записей Top-K в наборе данных

Question

В попытке выучить Hadoop я практикую нерешенные вопросы программирования из книги «Hadoop в действии»

Образец набора данных:

3070801,1963, 1096 ,, "BE", "" ,, 1,, 269,6,69,, 1,, 0 ,,,,,,, 3070802,1963,1096 ,, "US", "TX" ,, 1,, 2,6,63,, 0 ,,,,,,,,, 3070803,1963,1096 ,, "US", "IL" ,, 1,, 2,6,63,, 9, 0,3704,,,,,,, 3070804,1963,1096 ,, "US", "OH" ,, 1,, 2,6,63,, 3, 0,6667 ,,,,,,, 3070805,1963,1096 ,,"US", "CA" ,, 1,, 2,6,63,, 1,, 0 ,,,,,,, 3070806,1963,1096 ,, "US", "PA" ,, 1,, 2, 6,63,, 0 ,,,,,,,,, 3070807,1963,1096 ,, "US", "OH" ,, 1,, 623,3,39,, 3,, 0,4444 ,,,,,,, 3070808,1963,1096 ,, "US", "IA" ,, 1, 623,3,39,, 4, 0,375 ,,,,,,, 3070809,1963,1096 ,, "US", "AZ" ,, 1,, 4,6,65,, 0 ,,,,,,,,, 3070810,1963,1096 ,, "US", "IL" ,, 1, 4,6,65,, 3,, 0,4444 ,,,,,,,

Функция карты

public static class MapClass extends MapReduceBase implements Mapper<Text, Text, IntWritable, Text> {
        private int maxClaimCount = 0;
        private Text record = new Text();

        public void map(Text key, Text value, OutputCollector<IntWritable, Text> output, Reporter reporter) throws IOException {
            String claim = value.toString().split(",")[7];
            //if (!claim.isEmpty() && claim.matches("\\d")) {
            if (!claim.isEmpty()) {
                int claimCount = Integer.parseInt(claim);
                if (claimCount > maxClaimCount) {
                    maxClaimCount = claimCount;
                    record = value;
                    output.collect(new IntWritable(claimCount), value);
                }
//              output.collect(new IntWritable(claimCount), value);
            }
        }

    }

Функция уменьшения

public static class Reduce extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text> {

    public void reduce(IntWritable key, Iterator<Text> values, OutputCollector<IntWritable, Text> output, Reporter reporter) throws IOException {
            output.collect(key, values.next()); 
    }
}

Команда для запуска:

hadoop jar ~/Desktop/wc.jar com/hadoop/patent/TopKRecords -Dmapred.map.tasks=7 ~/input  ~/output

Требование:
- На основе значения в девятом столбце найдите записи верхнего K(скажем, 7) из набора данных

Вопрос:
- Поскольку требуется всего 7 верхних записей, я запускаю семь задач карты и проверяю, получаю ли запись с наибольшим номером как maxClaimCount и record
-не знаете, как собрать только максимальную запись, чтобы каждая карта выдавала только один вывод

Как мне это сделать?

Answer 1

Это обновленный ответ. Все комментарии к нему не применимы, так как основаны на оригинальном (неправильном) ответе.

---

Mapper должен выводить только

output.collect(new IntWritable(claimCount), value);

без сравнения. Результат будет отсортирован по количеству претензий и передан в редуктор.

В Редукторе используйте некоторую приоритетную очередь для получения 7 лучших результатов.

Answer 2

Вы можете использовать шаблоны топ-k для получения более подробной информации см. Ниже блог Поиск топ-K записей в Mapreduce

Answer 3

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

public Mapper() {
   TreeMap<String, String> set = new TreeMap<String, String>();
   Void map(object key, Text value){
     Set.put("get key", value);
     If(set.size > 7) {
        Set.removeFirst()
     }

   }

Public void cleanup(){
  While(Entry<string, string> entry : map.entrySet()) {
    Conetext.write(entry.key, entry.value);
  }
}}