hadoop中map/reduce数据求平均值实战

  • hadoop中map/reduce数据求平均值实战已关闭评论
  • 99 views
  • A+
所属分类:hadoop

说明

对输入文件中数据进行就算学生平均成绩。输入文件中的每行内容均为一个学生的姓名和他相应的成绩,如果有多门学科,则每门学科为一个文件。对输入文件中数据进行就算学生平均成绩。输入文件中的每行内容均为一个学生的姓名和他相应的成绩,如果有多门学科,则每门学科为一个文件。 要求在输出中每行有两个间隔的数据,其中,第一个代表学生的姓名,第二个代表其平均成绩。

设计思考

Map处理的 是一个纯文本文件, 文件中存放的数据时每一行表示一个学生的姓名和他相应一科成绩。Mapper处理的数据是由InputFormat分解过的数据集,Map处理的 是一个纯文本文件, 文件中存放的数据时每一行表示一个学生的姓名和他相应一科成绩。Mapper处理的数据是由InputFormat分解过的数据集, 其中 InputFormat的作用是将数据集切割成小数据集InputSplit,每一个InputSplit将由一个Mapper负责处理。此 外,InputFormat中还提供了一个RecordReader的实现, 并将一个InputSplit解析成<key,value>对提 供给了map函数。InputFormat的默认值是TextInputFormat,它针对文本文件,按行将文本切割成InputSlit, 并用 LineRecordReader将InputSplit解析成<key,value>对,key是行在文本中的位置,value是文件中的 一行。
Map的结果会通过partion分发到Reducer,Reducer做完Reduce操作后,将通过以格式OutputFormat输出。
Mapper最终处理的结果对<key,value>,会送到Reducer中进行合并,合并的时候,有相同key的键/值对则送到同一个 Reducer上。 Reducer是所有用户定制Reducer类地基础,它的输入是key和这个key对应的所有value的一个迭代器,同时还有 Reducer的上下文。 Reduce的结果由Reducer.Context的write方法输出到文件中。

原始数据

math:

 

张三 88

李四 99

王五 66

赵六 77

 

china:

 

张三 78

李四 89

王五 96

赵六 67

 

english:

 

张三 80

李四 82

王五 84

赵六 86

 

样本输出

张三 82

李四 90

王五 82

赵六 76

map代码

package com.test.hadoop.avg;

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

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class AvgScoreMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value,Context context)
            throws IOException, InterruptedException {
        String val = value.toString();
        StringTokenizer stringTokenizer = new StringTokenizer(val,"\n");
        while(stringTokenizer.hasMoreElements()){
            StringTokenizer tmp = new StringTokenizer(stringTokenizer.nextToken());
            String username = tmp.nextToken();
            String score = tmp.nextToken();

            context.write(new Text(username), new IntWritable(Integer.valueOf(score)));
        }

    }

}

reduce代码

package com.test.hadoop.avg;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class AvgScoreReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,Context context)
            throws IOException, InterruptedException {

        Iterator<IntWritable> iterator = values.iterator();
        int count = 0;
        int sum =0;
        while(iterator.hasNext()){
            int v = iterator.next().get();
            sum += v;
            count++;
        }
        int avg = sum/count;
        context.write(key, new IntWritable(avg));

    }


}

main代码

package com.test.hadoop.avg;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.wy.hadoop.join.two.UserJob;

public class AvgScoreJob extends Configuration implements Tool, Runnable {

    private String inputPath = null;
    private String outputPath = null;

    public AvgScoreJob(String inputPath,String outputPath){
        this.inputPath = inputPath;
        this.outputPath = outputPath;
    }
    public AvgScoreJob(){}

    @Override
    public Configuration getConf() {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public void setConf(Configuration arg0) {
        // TODO Auto-generated method stub

    }

    @Override
    public void run() {
        try{
            String[] args = {this.inputPath,this.outputPath};

            start(args);

        }catch (Exception e) {
            e.printStackTrace();
        }

    }

    private void start(String[] args)throws Exception{

        ToolRunner.run(new UserJob(), args);
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(configuration);
        fs.delete(new Path(args[1]),true);

        Job job = new Job(configuration,"avgjob");
        job.setJarByClass(AvgScoreJob.class);

        job.setInputFormatClass(TextInputFormat.class);

        job.setMapperClass(AvgScoreMapper.class);
        job.setReducerClass(AvgScoreReducer.class);

        job.setOutputFormatClass(TextOutputFormat.class);

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

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


        boolean success = job.waitForCompletion(true);
        return success?0:1;
    }

}


   package com.test.hadoop.avg;

public class JobMain {

    /**
     * @param args
     */
    public static void main(String[] args) {
        if(args.length==2){
            new Thread(new AvgScoreJob(args[0],args[1])).start();
        }

    }

}

1111111111111111111111111111111

  • 安卓客户端下载
  • 微信扫一扫
  • weinxin
  • 微信公众号
  • 微信公众号扫一扫
  • weinxin
avatar