MapReduce 計數器

1 篇文章 / 0 new

發表回應前，請先登入

週四, 2013-12-19 11:42

author

MapReduce 計數器

MR內建些計數器, 用來紀錄一些運作過程中的資訊提供給我們作參考, 同樣我們可在 mapper, reducer 內加入自己的計數器來紀錄我們需要的資訊以便於相關應用.
► 紀錄儲存(方式有兩種)

public class MaxMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    enum YearValue {//可藉由 MaxMapper_YearValue.properties 檔來置換成易讀的文字
        MAX, MIN
    }    
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        context.getCounter("Year",year).increment(1);//計數器,方式一
        context.getCounter(YearValue.MAX).increment(1);//計數器,方式二 enum型態
        ...
    }
}

► Enum keyName 的註解

若使用 enum 型態來當key進行計數器紀錄, 因會以 MAX=5 方式呈現, 若想更親和或多語系則可配置資源檔來進行自動轉換, 資源檔的命名規則為

類別名稱_ENUM名稱.properties, 如 MaxMapper_YearValue.properties
配合語系則為 類別名稱_ENUM名稱_語系代號.properties, 如 MaxMapper_YearValue_zh_TW.properties
資源檔資料內容為

CounterGroupName=年度最大值
MAX.name=最大值
MIN.name=最小值

則原 MAX=5 將變成 最大值=5 方式呈現

若使用 context.getCounter(String,String).increment(1) 方式紀錄則無此自動轉換機制.

► 計數器資料讀取

Driver 當 job 任務完成後立即讀取

int state = job.waitForCompletion(true) ? 0 : 1;
if (state==0) {
    Counters counters = job.getCounters();//取得計數器
    Counter counter = counters.findCounter(MaxMapper.YearValue.MAX);//查找计数器
    System.out.printf("%s:%d\n",counter.getDisplayName(), counter.getValue());
}
return state;

由 JobID 來讀取歷史紀錄

public class HistoryJobCounter extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        JobID jId = JobID.forName("job_201312190905_0001");
        if (jId == null) {
            System.err.printf("Job Id 不存在.\n");
            return -1;
        }
        JobClient jobClient = new JobClient(new JobConf(getConf()));
        RunningJob job = jobClient.getJob(jId);
        if (!job.isComplete()) {
            System.err.printf("任務[%s] 仍在執行中\n", job.getID());
            return -1;
        }
        //列出所有計數器
        for (String groupName : job.getCounters().getGroupNames()) {
            System.out.printf("群組名稱:%s\n",groupName);
            int i = 0;
            for (Counter counter : counters.getGroup(groupName)) {
                i++;
                System.out.printf(" [%02d] %s=%s\n KeyName:%s\n",i,counter.getDisplayName()
                        ,counter.getValue(), counter.getName());
            }
        }
        return 0;
    }
    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new HistoryJobCounter(), args));
    }
}

關鍵字:

MapReduce