Доступ к счетчику картографа от редуктора

Мне нужно получить доступ к счетчикам из моего картографа в моем редукторе. Это возможно? Если да, то как это делается?

В качестве примера: мой картограф:

public class CounterMapper extends Mapper<Text,Text,Text,Text> {

    static enum TestCounters { TEST }

    @Override
    protected void map(Text key, Text value, Context context)
                    throws IOException, InterruptedException {
        context.getCounter(TestCounters.TEST).increment(1);
        context.write(key, value);
    }
}

Мой редуктор

public class CounterReducer extends Reducer<Text,Text,Text,LongWritable> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
                        throws IOException, InterruptedException {
        Counter counter = context.getCounter(CounterMapper.TestCounters.TEST);
        long counterValue = counter.getValue();
        context.write(key, new LongWritable(counterValue));
    }
}

counterValue всегда 0. Я делаю что-то неправильно или это просто невозможно?

Ответ 1

В Reducer configure (JobConf) вы можете использовать объект JobConf для поиска идентификатора задания редуктора. При этом ваш редуктор может создать свой собственный JobClient - то есть соединение с jobtracker - и запросить счетчики для этого задания (или любое задание, если на то пошло).

// in the Reducer class...
private long mapperCounter;

@Override
public void configure(JobConf conf) {
    JobClient client = new JobClient(conf);
    RunningJob parentJob = 
        client.getJob(JobID.forName( conf.get("mapred.job.id") ));
    mapperCounter = parentJob.getCounters().getCounter(MAP_COUNTER_NAME);
}

Теперь вы можете использовать mapperCounter внутри самого метода reduce().

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

Обратите внимание, что счетчики mappers должны быть доработаны до того, как начнется какой-либо редуктор, поэтому, вопреки комментарию Джастина Томаса, я считаю, что вы должны получать точные значения (пока редукторы не увеличивают один и тот же счетчик!)

Ответ 2

Реализовано решение Jeff G для нового API:

    @Override
    public void setup(Context context) throws IOException, InterruptedException{
        Configuration conf = context.getConfiguration();
        Cluster cluster = new Cluster(conf);
        Job currentJob = cluster.getJob(context.getJobID());
        mapperCounter = currentJob.getCounters().findCounter(COUNTER_NAME).getValue();  
    }

Ответ 3

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

У них есть пример подсчета слов:

http://wiki.apache.org/hadoop/WordCount

Вы можете изменить context.write(word, one) на context.write(line, one)

Ответ 4

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

Ответ 5

Я спросил этот вопрос, но я не решил свою проблему. Однако альтернативное решение пришло мне в голову. В mapper подсчитывается количество слов, и оно может быть записано на промежуточный выход с минимальным ключом (так, чтобы это значение находилось в голове) в функции очистки, которая заканчивается концом mapper. В редукторе количество слов вычисляется путем добавления значений в голову. Пример кода и часть его вывода доступны ниже.

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by tolga on 1/26/16.
 */
public class WordCount {
    static enum TestCounters { TEST }
    public static class Map extends Mapper<Object, Text, Text, LongWritable> {
        private final static LongWritable one = new LongWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while (tokenizer.hasMoreTokens()) {
                word.set(tokenizer.nextToken());
                context.write(word, one);
                context.getCounter(TestCounters.TEST).increment(1);
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(new Text("!"),new LongWritable(context.getCounter(TestCounters.TEST).getValue()));
        }
    }

    public static class Reduce extends Reducer<Text, LongWritable, Text, LongWritable> {

        public void reduce(Text key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (LongWritable val : values) {
                sum += val.get();
            }
            context.write(key, new LongWritable(sum));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        Job job = new Job(conf, "WordCount");
        job.setJarByClass(WordCount.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.waitForCompletion(true);
    }
}

Текстовый файл:

Turgut Özal University is a private university located in Ankara, Turkey. It was established in 2008 by the Turgut Özal Thought and Action Foundation and is named after former Turkish president Turgut Özal.

Промежуточный выход

**!	33**
2008	1
Action	1
Ankara,	1
Foundation	1
It	1
Thought	1
Turgut	1
Turgut	1
Turgut	1

Ответ 6

Улучшение от ответа ицхаки

findCounter(COUNTER_NAME) больше не поддерживается - https://hadoop.apache.org/docs/r2.7.0/api/org/apache/hadoop/mapred/Counters.html

@Override
public void setup(Context context) throws IOException, InterruptedException{
    Configuration conf = context.getConfiguration();
    Cluster cluster = new Cluster(conf);
    Job currentJob = cluster.getJob(context.getJobID());
    mapperCounter = currentJob.getCounters().findCounter(GROUP_NAME, COUNTER_NAME).getValue();  
}

GROUP_NAME указывается при вызове счетчика. например.

context.getCounter("com.example.mycode", "MY_COUNTER").increment(1);

затем

mapperCounter = currentJob.getCounters().findCounter("com.example.mycode", "MY_COUNTER").getValue();

Также важно отметить, что если счетчик не существует, он инициализирует значение со значением 0.