Hadoop MapReduce编程 API入门系列之MapReduce多种输入格式（十七）

　　不多说，直接上代码。

 

代码

package zhouls.bigdata.myMapReduce.ScoreCount;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
/**
* 学习成绩读写类
* 数据格式参考：19020090017 小讲 90 99 100 89 95
* @author Bertron
* 需要自定义一个 ScoreWritable 类实现 WritableComparable 接口，将学生各门成绩封装起来。
*/
public class ScoreWritable implements WritableComparable< Object > {//其实这里，跟TVPlayData一样的
//  注意：    Hadoop通过Writable接口实现的序列化机制，不过没有提供比较功能，所以和java中的Comparable接口合并，提供一个接口WritableComparable。（自定义比较）
//         Writable接口提供两个方法(write和readFields)。

    
    private float Chinese;
    private float Math;
    private float English;
    private float Physics;
    private float Chemistry;
    
    
//    问：这里我们自己编程时，是一定要创建一个带有参的构造方法，为什么还要显式的写出来一个带无参的构造方法呢？
//    答：构造器其实就是构造对象实例的方法，无参数的构造方法是默认的，但是如果你创造了一个带有参数的构造方法，那么无参的构造方法必须显式的写出来，否则会编译失败。
    
    public ScoreWritable(){}//java里的无参构造函数，是用来在创建对象时初始化对象  
    //在hadoop的每个自定义类型代码里，好比，现在的ScoreWritable，都必须要写无参构造函数。
    
    
    //问：为什么我们在编程的时候，需要创建一个带有参的构造方法？
    //答:就是能让赋值更灵活。构造一般就是初始化数值，你不想别人用你这个类的时候每次实例化都能用另一个构造动态初始化一些信息么(当然没有需要额外赋值就用默认的)。
    
    public ScoreWritable(float Chinese,float Math,float English,float Physics,float Chemistry){//java里的有参构造函数,是用来在创建对象时初始化对象  
        this.Chinese = Chinese;
        this.Math = Math;
        this.English = English;
        this.Physics = Physics;
        this.Chemistry = Chemistry;
    }
    
    //问：其实set和get方法，这两个方法只是类中的setxxx和getxxx方法的总称，
    //    那么，为什么在编程时，有set和set***两个，只有get***一个呢？
    
    public void set(float Chinese,float Math,float English,float Physics,float Chemistry){
        this.Chinese = Chinese;//即float Chinese赋值给private float Chinese;
        this.Math = Math;
        this.English = English;
        this.Physics = Physics;
        this.Chemistry = Chemistry;
    }
//    public float get(float Chinese,float Math,float English,float Physics,float Chemistry){因为这是错误的，所以对于set可以分开，get只能是get***
//        return Chinese;
//        return Math;
//        return English;
//        return Physics;
//        return Chemistry;
//    }
    
    
    public float getChinese() {//拿值，得返回，所以需有返回类型float
        return Chinese;
    }
    public void setChinese(float Chinese){//设值，不需，所以空返回类型
        this.Chinese = Chinese;
    }
    public float getMath() {//拿值
        return Math;
    }
    public void setMath(float Math){//设值
        this.Math = Math;
    }
    public float getEnglish() {//拿值
        return English;
    }
    public void setEnglish(float English){//设值
        this.English = English;
    }
    public float getPhysics() {//拿值
        return Physics;
    }
    public void setPhysics(float Physics){//设值
        this.Physics = Physics;
    }
    public float getChemistry() {//拿值
        return Chemistry;
    }
    public void setChemistry(float Chemistry) {//拿值
        this.Chemistry = Chemistry;
    }
    
    // 实现WritableComparable的readFields()方法
//    对象不能传输的，需要转化成字节流！
//    将对象转换为字节流并写入到输出流out中是序列化，write 的过程（最好记!!!）
//    从输入流in中读取字节流反序列化为对象      是反序列化，readFields的过程（最好记!!!）
    public void readFields(DataInput in) throws IOException {//拿代码来说的话，对象就是比如Chinese、Math。。。。
        Chinese = in.readFloat();//因为，我们这里的对象是float类型，所以是readFloat()
        Math = in.readFloat();
        English = in.readFloat();//注意:反序列化里，需要生成对象对吧，所以，是用到的是get那边对象
        Physics = in.readFloat();
        Chemistry = in.readFloat();
//        in.readByte()
//        in.readChar()
//        in.readDouble()
//        in.readLine() 
//        in.readFloat()
//        in.readLong()
//        in.readShort()
    }
    
    // 实现WritableComparable的write()方法，以便该数据能被序列化后完成网络传输或文件输出 
//    将对象转换为字节流并写入到输出流out中是序列化，write 的过程（最好记!!!）
//    从输入流in中读取字节流反序列化为对象      是反序列化，readFields的过程（最好记!!!）
    public void write(DataOutput out) throws IOException {//拿代码来说的话，对象就是比如Chinese、Math。。。。
        out.writeFloat(Chinese);//因为，我们这里的对象是float类型，所以是writeFloat()
        out.writeFloat(Math);
        out.writeFloat(English);//注意:序列化里，需要对象对吧，所以，用到的是set那边的对象
        out.writeFloat(Physics);
        out.writeFloat(Chemistry);
//        out.writeByte()
//        out.writeChar()
//        out.writeDouble()
//        out.writeFloat()
//        out.writeLong()
//        out.writeShort()
//        out.writeUTF()
    }
    
    public int compareTo(Object o) {//java里的比较，Java String.compareTo()
        return 0;
    }
    
    
//    Hadoop中定义了两个序列化相关的接口：Writable 接口和 Comparable 接口，这两个接口可以合并成一个接口 WritableComparable。
//    Writable 接口中定义了两个方法，分别为write(DataOutput out)和readFields(DataInput in)
//    所有实现了Comparable接口的对象都可以和自身相同类型的对象比较大小
    
    
//    Hadoop中定义了两个序列化相关的接口：Writable 接口和 Comparable 接口，这两个接口可以合并成一个接口 WritableComparable。
//    Writable 接口中定义了两个方法，分别为write(DataOutput out)和readFields(DataInput in)
//    所有实现了Comparable接口的对象都可以和自身相同类型的对象比较大小
    
    
//  源码是
//    package java.lang;  
//    import java.util.*;      
//    public interface Comparable {  
//        /** 
//        * 将this对象和对象o进行比较，约定：返回负数为小于，零为大于，整数为大于 
//        */  
//        public int compareTo(T o);  
//    }
    
}

package zhouls.bigdata.myMapReduce.ScoreCount;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;
/**
* 自定义学生成绩读写InputFormat
* 数据格式参考：19020090017 小讲 90 99 100 89 95
* @author Bertron
*/

            //其实这个程序，就是在实现InputFormat接口，TVPlayInputFormat是InputFormat接口的实现类
            //比如   ScoreInputFormat  extends FileInputFormat implements InputFormat。

            //问：自定义输入格式 ScoreInputFormat 类，首先继承 FileInputFormat，然后分别重写 isSplitable() 方法和 createRecordReader() 方法。

public class ScoreInputFormat extends FileInputFormat<Text,ScoreWritable > {//自定义数据输入格式，其实这都是模仿源码的！可以去看

//    线路是： boolean  isSplitable()   ->   RecordReader<Text,ScoreWritable> createRecordReader()   ->   ScoreRecordReader extends RecordReader<Text, ScoreWritable > 
    
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {//这是InputFormat的isSplitable方法
            //isSplitable方法就是是否要切分文件，这个方法显示如果是压缩文件就不切分，非压缩文件就切分。
//        如果不允许分割，则isSplitable==false，则将第一个block、文件目录、开始位置为0，长度为整个文件的长度封装到一个InputSplit，加入splits中
//        如果文件长度不为0且支持分割，则isSplitable==true,获取block大小，默认是64MB
        return false;    //整个文件封装到一个InputSplit
        //要么就是return true;        //切分64MB大小的一块一块，再封装到InputSplit
    }
    
    @Override
    public RecordReader<Text,ScoreWritable> createRecordReader(InputSplit inputsplit,TaskAttemptContext context) throws IOException, InterruptedException {
//        RecordReader<k1, v1>是返回类型,返回的RecordReader对象的封装
//        createRecordReader是方法,在这里是，ScoreInputFormat.createRecordReader。ScoreInputFormat是InputFormat类的实例
//        InputSplit input和TaskAttemptContext context是传入参数
        
//        isSplitable(),如果是压缩文件就不切分，整个文件封装到一个InputSplit
//        isSplitable(),如果是非压缩文件就切，切分64MB大小的一块一块，再封装到InputSplit
        
        //这里默认是系统实现的的RecordReader，按行读取，下面我们自定义这个类ScoreRecordReader。
        //类似与Excel、WeiBo、TVPlayData代码写法
        return new ScoreRecordReader();//新建一个ScoreRecordReader实例，所有才有了上面RecordReader<Text,ScoreWritable>，所以才如下ScoreRecordReader，写我们自己的
    }
    
    
    //RecordReader中的两个参数分别填写我们期望返回的key/value类型，我们期望key为Text类型，value为ScoreWritable类型封装学生所有成绩
    public static class ScoreRecordReader extends RecordReader<Text, ScoreWritable > {//RecordReader<k1, v1>是一个整体
        public LineReader in;//行读取器
        public Text line;//每行数据类型
        public Text lineKey;//自定义key类型，即k1
        public ScoreWritable lineValue;//自定义value类型，即v1
        
        @Override
        public void close() throws IOException {//关闭输入流
            if(in !=null){
                in.close();
            }
        }
        @Override
        public Text getCurrentKey() throws IOException, InterruptedException {//获取当前的key,即CurrentKey
            return lineKey;//返回类型是Text,即Text lineKey
        }
        @Override
        public ScoreWritable getCurrentValue() throws IOException,InterruptedException {//获取当前的Value，即CurrentValue
            return lineValue;//返回类型是ScoreWritable,即ScoreWritable lineValue
        }
        @Override
        public float getProgress() throws IOException, InterruptedException {//获取进程，即Progress
            return 0;//返回类型是float,即float 0
        }
        @Override
        public void initialize(InputSplit input, TaskAttemptContext context) throws IOException, InterruptedException {//初始化，都是模板
            FileSplit split=(FileSplit)input;
            Configuration job=context.getConfiguration();
            Path file=split.getPath();
            FileSystem fs=file.getFileSystem(job);
            
            FSDataInputStream filein=fs.open(file);
            in=new LineReader(filein,job);//输入流in
            line=new Text();//每行数据类型
            lineKey=new Text();//自定义key类型，即k1。//新建一个Text实例作为自定义格式输入的key
            lineValue = new ScoreWritable();//自定义value类型，即v1。//新建一个TVPlayData实例作为自定义格式输入的value
        }
        
        //此方法读取每行数据，完成自定义的key和value
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {//这里面，才是篡改的重点
            int linesize=in.readLine(line);//line是每行数据，我们这里用到的是in.readLine(str)这个构造函数，默认读完读到文件末尾。其实这里有三种。
            
//            是SplitLineReader.readLine  ->  SplitLineReader  extends   LineReader  ->  org.apache.hadoop.util.LineReader
            
//            in.readLine(str)//这个构造方法执行时，会首先将value原来的值清空。默认读完读到文件末尾
//            in.readLine(str, maxLineLength)//只读到maxLineLength行
//            in.readLine(str, maxLineLength, maxBytesToConsume)//这个构造方法来实现不清空，前面读取的行的值

            if(linesize==0) return false;
            
            
            String[] pieces = line.toString().split("\s+");//解析每行数据
                    //因为，我们这里是。默认读完读到文件末尾。line是Text类型。pieces是String[]，即String数组。
            
            if(pieces.length != 7){
                throw new IOException("Invalid record received");
            }
            //将学生的每门成绩转换为 float 类型
            float a,b,c,d,e;
            try{
                a = Float.parseFloat(pieces[2].trim());//将String类型，如pieces[2]转换成，float类型，给a
                b = Float.parseFloat(pieces[3].trim());
                c = Float.parseFloat(pieces[4].trim());
                d = Float.parseFloat(pieces[5].trim());
                e = Float.parseFloat(pieces[6].trim());
            }catch(NumberFormatException nfe){
                throw new IOException("Error parsing floating poing value in record");
            }
            lineKey.set(pieces[0]+"	"+pieces[1]);//完成自定义key数据
            lineValue.set(a, b, c, d, e);//封装自定义value数据
//            或者写
//            lineValue.set(Float.parseFloat(pieces[2].trim()),Float.parseFloat(pieces[3].trim()),Float.parseFloat(pieces[4].trim()),
//                    Float.parseFloat(pieces[5].trim()),Float.parseFloat(pieces[6].trim()));
            
//            pieces[0]   pieces[1] pieces[2]  ... pieces[6]
//            19020090040 秦心芯 123 131 100 95 100
//            19020090006 李磊 99 92 100 90 100
//            19020090017 唐一建 90 99 100 89 95
//            19020090031 曾丽丽 100 99 97 79 96
//            19020090013 罗开俊 105 115 94 45 100
//            19020090039 周世海 114 116 93 31 97
//            19020090020 王正伟 109 98 88 47 99
//            19020090025 谢瑞彬 94 120 100 50 73
//            19020090007 于微 89 78 100 66 99
//            19020090012 刘小利 87 82 89 71 99
            
            
            
            return true;
        }        
    }
}

package zhouls.bigdata.myMapReduce.ScoreCount;


import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* 学生成绩统计Hadoop程序
* 数据格式参考：19020090017 小讲 90 99 100 89 95
* @author HuangBQ
*/
public class ScoreCount extends Configured implements Tool{
    public static class ScoreMapper extends Mapper<Text,ScoreWritable,Text,ScoreWritable>{
        @Override
        protected void map(Text key, ScoreWritable value, Context context)throws IOException, InterruptedException{
            context.write(key, value);//写入key是k2,value是v2
//            context.write(new Text(key), new ScoreWritable(value));等价           
        }
    }
    
    public static class ScoreReducer extends Reducer<Text,ScoreWritable,Text,Text>{
        private Text text = new Text();
        protected void reduce(Text Key, Iterable< ScoreWritable > Values, Context context)throws IOException, InterruptedException{
            float totalScore=0.0f;
            float averageScore = 0.0f;
            for(ScoreWritable ss:Values){
                totalScore +=ss.getChinese()+ss.getMath()+ss.getEnglish()+ss.getPhysics()+ss.getChemistry();
                averageScore +=totalScore/5;
            }
            text.set(totalScore+"	"+averageScore);
            context.write(Key, text);//写入Key是k3,text是v3
//            context.write(new Text(Key),new Text(text));等价            
        }
    }
    

    public int run(String[] args) throws Exception{
        Configuration conf = new Configuration();//读取配置文件
        
        Path mypath = new Path(args[1]);
        FileSystem hdfs = mypath.getFileSystem(conf);//创建输出路径
        if (hdfs.isDirectory(mypath)) 
        {
            hdfs.delete(mypath, true);
        }
        
        Job job = new Job(conf, "ScoreCount");//新建任务
        job.setJarByClass(ScoreCount.class);//设置主类
        
        FileInputFormat.addInputPath(job, new Path(args[0]));// 输入路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));// 输出路径
        
        job.setMapperClass(ScoreMapper.class);// Mapper
        job.setReducerClass(ScoreReducer.class);// Reducer
        
        job.setMapOutputKeyClass(Text.class);// Mapper key输出类型
        job.setMapOutputValueClass(ScoreWritable.class);// Mapper value输出类型
                
        job.setInputFormatClass(ScoreInputFormat.class);//设置自定义输入格式
        
        job.waitForCompletion(true);        
        return 0;
    }
    
    
    
    public static void main(String[] args) throws Exception{
//        String[] args0 = 
//                { 
//                "hdfs://HadoopMaster:9000/score/score.txt",
//                "hdfs://HadoopMaster:9000/out/score/" 
//                };
        
        String[] args0 = 
            { 
            "./data/score/score.txt",
            "./out/score/" 
            };
        
        int ec = ToolRunner.run(new Configuration(), new ScoreCount(), args0);
        System.exit(ec);
    }
}