Hadoop工程师面试题(1)--MapReduce实现单表汇总统计

数据源格式描述:

输入t1.txt源数据,数据文件分隔符”*&*”,字段说明如下:

字段序号 字段英文名称 字段中文名称 字段类型 字段长度
1 TIME_ID 时间(到时) 字符型 12
2 Session 会话时长 数值型 8
3 MSISDN 用户号码 字符型 11
4 SP_DOMAIN SP域名 数值型 64
5 USER_AGENT_ORIGN 终端字串 字符型 128
6 USER_AGENT 终端类别 字符型 64
7 UPSTREAM_VOL 上行流量 数值型 8
8 DOWNSTREAM_VOL 下行流量 数值型 8
9 URL_CNT 访问次数 数值型 20

用mapreduce实现单表汇总:

在数据源的基础上,根据终端类型汇总出总流量及访问次数。汇总模型字段说明如下:

字段序号 字段英文名称 字段中文名称 字段类型 字段长度
1 USER_AGENT 终端类型 字符型
2 TOT_FLUX 总流量 数值型 30
3 URL_CNT 访问次数 数值型 30

代码如下:

package mianshi;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
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 org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;

import com.google.protobuf.TextFormat;

public class Test1 {

/**
     * @param args
     * @throws IOException
     * @throws InterruptedException
     * @throws ClassNotFoundException
     */
    public static void main(String[] args) throws Exception {
        //创建配置文件
        Configuration conf=new Configuration();
        //创建job
        Job job = new Job(conf,Test1.class.getName());
        //设置jar包运行
        job.setJarByClass(Test1.class);
        //设置输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //设置输入格式
        job.setInputFormatClass(TextInputFormat.class);
        //设置自定义Mapper
        job.setMapperClass(MyMapper.class);
        //设置Map输出的Value类型,也就是V2
        job.setMapOutputValueClass(Model.class);
        //设置Map输出的Key类型,也就是K2
        job.setMapOutputKeyClass(Text.class);
        //设置分区类型
        job.setPartitionerClass(HashPartitioner.class);
        //设置Rudece任务数
        job.setNumReduceTasks(1);
        //设置自定义Reduce类
        job.setReducerClass(MyReducer.class);
        //设置输出K3的类型
        job.setOutputKeyClass(Text.class);
        //设置输出的V3类型
        job.setOutputValueClass(Model.class);
        //设置输出的格式
        job.setOutputFormatClass(TextOutputFormat.class);
        //指定输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        //提交job
        job.waitForCompletion(true);

}
    static class MyMapper extends Mapper<LongWritable, Text, Text, Model>{
        @Override
        protected void map(LongWritable k1, Text v1,Context context)
                throws IOException, InterruptedException {
            /**
             * 切割字符串有点意思!
             * “*”是特殊字符,需要用[]
             * "&"需要用\\转义
             *
             *    
             */
            String[] split = v1.toString().split("[*]\\&[*]");
            Text user_agent = new Text(split[5]);
            Long tot_flux = new Long(split[6])+new Long(split[7]);
            Long url_cnt = new Long(split[8]);
            Model v2 = new Model(tot_flux, url_cnt);
            context.write(user_agent, v2);
           
        }
    }
   
    static class MyReducer extends Reducer<Text, Model, Text, Model>{
       
        @Override
        protected void reduce(Text k2, Iterable<Model> v2s,Context context)
                throws IOException, InterruptedException {
           
            //定义计数器
            long sum_flux =0L;
            long sum_url = 0L;
            for(Model model : v2s){
                sum_flux += model.tot_flux;
                sum_url += model.url_cnt;
            }
            Model v3 = new Model(sum_flux,sum_url);
            context.write(k2, v3);
        }
       
    }

}

/**
* 自定义类型必须实现Writable
* @author Sky
*
*/
class Model implements Writable{
   
    long tot_flux;
    long url_cnt;
   
    public Model(){}
    public Model(Long tot_flux,Long url_cnt){
        this.tot_flux = tot_flux;
        this.url_cnt = url_cnt;
    }

public void write(DataOutput out) throws IOException {
        //序列化出去
        out.writeLong(tot_flux);
        out.writeLong(url_cnt);
    }

public void readFields(DataInput in) throws IOException {
        //和序列化出去的一样
        this.tot_flux = in.readLong();
        this.url_cnt = in.readLong();
       
    }
   
    //必须覆写toString方法,否则输出的值是内存值
    @Override
    public String toString() {
        return tot_flux+"\t"+url_cnt;
    }
   
   
}

文章参考论坛:超人hadoop网络学院论坛

上一篇:hibernate缓存机制详细分析 复制代码 内部资料 请勿转载 谢谢合作


下一篇:hadoop-1.2.0安装记录