使用MapReduce运行自定义bean案例

傷城~ 2022-11-14 13:26 203阅读 0赞

用户编写的程序分成三个部分：Mapper、Reducer 和 Driver。

文件内容如下

手机号、IP、访问网站、上行流量、下行流量、状态码

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNDUwMzUyNg_size_16_color_FFFFFF_t_70][]

需求：需要统计每个手机号访问网站的上行流量、下行流量以及它们的总和

依赖包

>     <dependency>
>         <groupId>org.apache.hadoop</groupId>
>         <artifactId>hadoop-client</artifactId>
>         <version>3.1.3</version>
>     </dependency>
>     
>     <dependency>
>         <groupId>org.projectlombok</groupId>
>         <artifactId>lombok</artifactId>
>         <optional>true</optional>
>     </dependency>

## 1.创建实体FlowBean 实现Writable 序列化 ##

在Hadoop中，如果有Reduce阶段。通常key-value都需要实现序列化协议！
    MapTask处理后的key-value，只是一个阶段性的结果！
    这些key-value需要传输到ReduceTask所在的机器！
    将一个对象通过序列化技术，序列化到一个文件中，经过网络传输到另外一台机器，
    再使用反序列化技术，从文件中读取数据，还原为对象是最快捷的方式！

java的序列化协议： Serializable  
特点：不仅保存对象的属性值，类型，还会保存大量的包的结构，子父类和接口的继承信息，很笨重。  
hadoop开发了一款轻量级的序列化协议： Writable机制！

序列化是什么  
序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储（持久化）和网络传输。  
反序列化就是将收到字节序列（或其他数据传输协议）或者是硬盘的持久化数据，转换成内存中的对象。  
为什么要序列化  
一般来说，“活的”对象只生存在内存里，关机断电就没有了。  
而且“活的”对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。

>     import lombok.Data;
>     import org.apache.hadoop.io.Writable;
>     
>     import java.io.DataInput;
>     import java.io.DataOutput;
>     import java.io.IOException;
>     
>     /**
>      * FlowBean 序列化
>      *
>      * @author
>      * @description:
>      * @time: 2021/3/24 16:12
>      */
>     @Data
>     public class FlowBean implements Writable {
>          /**
>          *  上行流量
>          */
>         private long upFlow;
>         /**
>          * 下行流量
>          */
>         private long downFlow;
>         /**
>          * 总流量
>          */
>         private long sumFlow;
>     
>         public FlowBean() {
>     
>         }
>     
>         /**
>          * 序列化
>          *
>          * @param out
>          * @throws IOException
>          */
>         @Override
>         public void write(DataOutput out) throws IOException {
>             out.writeLong(upFlow);
>             out.writeLong(downFlow);
>             out.writeLong(sumFlow);
>         }
>     
>         /**
>          * 反序列化
>          * 顺序必须保持一致
>          *
>          * @param in
>          * @throws IOException
>          */
>         @Override
>         public void readFields(DataInput in) throws IOException {
>             upFlow = in.readLong();
>             downFlow = in.readLong();
>             sumFlow = in.readLong();
>         }
>     
>         @Override
>         public String toString() {
>             return upFlow + "\t" + downFlow + "\t" + sumFlow;
>         }
>     
>     }

## 2.创建 FlowBeanMapper ##

>     import org.apache.hadoop.io.LongWritable;
>     import org.apache.hadoop.io.Text;
>     import org.apache.hadoop.mapreduce.Mapper;
>     
>     import java.io.IOException;
>     
>     public class FlowBeanMapper extends Mapper<LongWritable, Text, Text, FlowBean> {
>     
>         private Text outK = new Text();
>         private FlowBean outV = new FlowBean();
>     
>         @Override
>         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
>     
>             // 1 获取一行
>     
>             String line = value.toString();
>     
>             // 2 切割
>      
>             // 2   13846544121    192.196.100.2        264    0  200  6 - 3 = 3
>             String[] split = line.split("\t");
>     
>             // 3 抓取想要的数据
>             // 手机号：13736230513
>             // 上行流量和下行流量：2481,24681
>             String phone = split[1];
>             String up = split[split.length - 3];
>             String down  = split[split.length - 2];
>     
>             // 4封装
>             outK.set(phone);
>             outV.setUpFlow(Long.parseLong(up));
>             outV.setDownFlow(Long.parseLong(down));
>             outV.setSumFlow(Long.parseLong(up)+Long.parseLong(down));
>     
>             // 5 写出
>             context.write(outK, outV);
>         }
>     }

##  3.创建FlowBeanReducer ##

>     import org.apache.hadoop.io.Text;
>     import org.apache.hadoop.mapreduce.Reducer;
>     
>     import java.io.IOException;
>     
>     public class FlowBeanReducer extends Reducer<Text, FlowBean,Text, FlowBean> {
>         private FlowBean outV = new FlowBean();
>     
>         @Override
>         protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
>     
>             // 1 遍历集合累加值
>             long sumUpFlow=0;
>             long sumDownFlow=0;
>     
>             for (FlowBean flowBean : values) {
>     
>                 sumUpFlow+=flowBean.getUpFlow();
>                 sumDownFlow+=flowBean.getDownFlow();
>     
>             }
>     
>             out_value.setUpFlow(sumUpFlow);
>             out_value.setDownFlow(sumDownFlow);
>             out_value.setSumFlow(sumDownFlow+sumUpFlow);
>     
>             context.write(key, out_value);
>     }

## 4.创建FlowDriver ##

> public class FlowBeanDriver \{
> 
>     public static void main(String\[\] args) throws Exception \{
> 
>         Path inputPath=new Path("e:/mrinput/flowbean");  
>         Path outputPath=new Path("e:/mroutput/flowbean");
> 
>         //作为整个Job的配置  
>         Configuration conf = new Configuration();
> 
>   
>         FileSystem fs=FileSystem.get(conf);
> 
>         if (fs.exists(outputPath)) \{          
>             fs.delete(outputPath, true);//保证输出目录不存在  
>         \}
> 
>         // ①创建Job  
>         Job job = Job.getInstance(conf);
> 
>         // ②设置Job  
>         // 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型  
>         job.setMapperClass(FlowBeanMapper.class);  
>         job.setReducerClass(FlowBeanReducer.class);
> 
>         // Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化  
>         // 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型  
>         job.setOutputKeyClass(Text.class);  
>         job.setOutputValueClass(FlowBean.class);
> 
>         // 设置输入目录和输出目录  
>         FileInputFormat.setInputPaths(job, inputPath);  
>         FileOutputFormat.setOutputPath(job, outputPath);
> 
>         // ③运行Job  
>         job.waitForCompletion(true);  
>     \}
> 
> \}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNDUwMzUyNg_size_16_color_FFFFFF_t_70]: /images/20221022/709ba0119550427abe5440589d5ba5a4.png