Hadoop之Mapreduce序列化怎么實現(xiàn)

這篇“Hadoop之Mapreduce序列化怎么實現(xiàn)”文章的知識點大部分人都不太理解，所以小編給大家總結(jié)了以下內(nèi)容，內(nèi)容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Hadoop之Mapreduce序列化怎么實現(xiàn)”文章吧。

什么是序列化：    

序列化就是把內(nèi)存中的對象，轉(zhuǎn)換成字節(jié)序列（或其他數(shù)據(jù)傳輸協(xié)議）以便于存儲到磁盤（持久化）和網(wǎng)絡傳輸。

什么是反序列化：

 反序列化就是將收到字節(jié)序列（或其他數(shù)據(jù)傳輸協(xié)議）或者是磁盤的持久化數(shù)據(jù)，轉(zhuǎn)換成內(nèi)存中的對象。

為什么要序列化：

一般來說，“活的”對象只生存在內(nèi)存里，關(guān)機斷 電就沒有了。而且“活的”對象只能由本地的進程使用，不能被發(fā)送到網(wǎng)絡上的另外一臺計算機。 然而序列化可以存儲“活的”對象，可以將“活的”對象發(fā)送到遠程計算機。

Java的序列化：

在Java中也是有序列化的，我們?yōu)槭裁床煌ㄟ^idea使用Java的序列化那？

因為Java的序列化框架（Serializable）是一個繁重的框架，附帶信息比較多（各種校驗信息，Header，繼承體系等），不便于在網(wǎng)絡中高效傳輸。所以，Hadoop自己開發(fā)了一套序列化機制（Writable）。

Hadoop序列化:

Hadoop的序列化比較精簡，只有簡單的校驗，有緊湊（高效使用存儲空間），快速（讀寫數(shù)據(jù)的額外開銷小），互操作（支持多語言的交互）的特點。

自定義序列化接口：  

在開發(fā)過程中，基本序列化類型不能滿足所有需求，比如在Hadoop框架內(nèi)部傳遞一個bean對象（不是基本的數(shù)據(jù)類型（某個類）----沒有對應的Hadoop類型），那么該對象就需要實現(xiàn)序列化接口。

實現(xiàn)序列化的步驟：

先看源碼進行簡單分析：

Writable接口（好像也分析不出什么）

兩個方法：

1.write: 進行序列化

2.readFields:進行反序列化 

 (1)  反序列化時，需要反射調(diào)用空參構(gòu)造函數(shù)，所以必須有空參構(gòu)造

public FlowBean() {
	super();
}

(2)  重寫接口中的兩個方法***(注意：反序列化的順序和序列化的順序完全一致)

                如數(shù)據(jù)結(jié)構(gòu)中的隊列一樣先進先出，先序列化則先反序列化

(3）需要重寫toString（）方法，因為需要打印出來,否則打印出來的是地址

(4)  如果需要將自定義的bean放在key中傳輸，則還需要實現(xiàn)Comparable接口，因為MapReduce框中的Shuffle過程要求對key必須能排序。(比如：上一篇博客中的計算單詞出現(xiàn)次數(shù)中 最后呈現(xiàn)的單詞是按照26個英文字母的順序進行排序的)

看一個樣例源碼（字符串Text）：

看到上圖  實現(xiàn)接口：

WritableComparable<BinaryComparable>

跟進一下：

 看到該接口繼承自Comparable接口（這是Java中的一個API）

序列化案例實操：

案例需求：

        統(tǒng)計每一個手機號耗費的總上行流量、總下行流量、總流量

（1）輸入數(shù)據(jù)：

（2）輸入數(shù)據(jù)格式：

（3）期望輸出數(shù)據(jù)格式

需求分析：

         先輸入數(shù)據(jù)，輸入數(shù)據(jù)后需要進行mapper階段---》reduce階段---》輸出階段

mapper階段：

先考慮輸入kv   （k---偏移量    v是一行數(shù)據(jù)）

輸出(kv)為reduce的輸入（kv） (本樣例中使用的k是手機號--統(tǒng)計手機號的流量使用      v為上行流量，下行流量，總流量    則需要封裝bean類（自定義對象）  再進行序列化傳輸（為什么要進行序列化那？----因為再計算的過程中可能由于資源問題mapper和reduce不在同一臺服務器上))

輸出（kv）同樣也是（手機號，bean類）

編寫MapperReduce程序：

1.FlowBean代碼：

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.Writable;
 
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
 
/*
*
* 定義bean類
* 需要實現(xiàn)writable
* 重寫序列化和反序列化方法
* 重寫空參構(gòu)造
* 重寫tostring方法
*
* */
public class FlowBean  implements Writable {
   private  long upFlow;
    private  long downFlow;
    private  long sumFlow;
 
    public long getUpFlow() {
        return upFlow;
    }
 
    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }
 
    public long getDownFlow() {
        return downFlow;
    }
 
    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }
 
    public long getSumFlow() {
        return sumFlow;
    }
 
    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }
    public void setSumFlow() {
        this.sumFlow = this.downFlow+this.upFlow;
    }
    //生成空參構(gòu)造函數(shù)由于反射  快捷鍵alt   + insert
 
    public FlowBean() {
    }
 
    @Override
    public void write(DataOutput out) throws IOException {
        //序列化方法
        //  向緩沖流中寫入Long類型的數(shù)據(jù)
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }
 
    @Override
    public void readFields(DataInput in) throws IOException {
//反序列化方法
        //讀取緩沖區(qū)中數(shù)據(jù)
        this.upFlow= in.readLong();
        this.downFlow= in.readLong();
        this.sumFlow= in.readLong();
    }
 
    @Override
    public String toString() {
        return upFlow + "\t"+downFlow +"\t"+ sumFlow ;
    }
}

2.Mapper代碼：

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
 
import java.io.IOException;
 
public class FlowMapper extends Mapper<LongWritable, Text,Text,FlowBean> {
 
    private Text outK=new Text();
    private  FlowBean outV=new FlowBean();  //調(diào)用的無參構(gòu)造函數(shù)
 
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        //1 獲取一行
        //1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
 
        String s = value.toString();// 將數(shù)據(jù)轉(zhuǎn)換成string
        //2 進行切割
 
        String[] split = s.split("\t"); //將數(shù)據(jù)按寫入形式進行切割
        //3 抓取想要的數(shù)據(jù)
        //根據(jù)角標獲取  手機號  上行流量  下行流量
 
 
        String phone = split[1];
        String up = split[split.length - 3];//  不能正序 因為有的屬性是沒有字段的
        String down = split[split.length - 2];
//     封裝輸出的kv
 
        outK.set(phone);
        outV.setUpFlow(Long.parseLong(up));//  up為string類型
        outV.setDownFlow(Long.parseLong(down));
        outV.setSumFlow();          //
 
        //寫出
        context.write(outK,outV);
    }
}

3. reduce代碼：

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
 
import java.io.IOException;
 
public class FlowReducer extends Reducer  <Text,FlowBean,Text,FlowBean>{
   private FlowBean outv=new FlowBean();
 
 
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
 
         long totalUp=0;
         long totaldown=0;
         
        //分析   傳入TEXT  為手機號  后邊為集合（Bean類的對象的集合）輸出還是一個一個bean類  （每個手機號的總和）
        for (FlowBean value : values) {  //傳入的參數(shù)是同一個key的
            totalUp+=value.getUpFlow();
            totaldown+=value.getDownFlow();
        }
        //  現(xiàn)在求出的是每個手機號的總的上行流量  下行流量
            //封裝  key不需要
        //outv
    outv.setUpFlow(totalUp);
    outv.setDownFlow(totaldown);
    outv.setSumFlow();
    //寫出
        context.write(key,outv);
    }
}

4.driver代碼：

package com.tangxiaocong.mapreduce.writable;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.output.FileOutputFormat;
 
import java.io.IOException;
 
public class FlowDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //獲取JOB
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);
 
        job.setJarByClass(FlowDriver.class);
        //關(guān)聯(lián)mapper  和reduce
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
 
        //設(shè)置mapper  輸出的key 和value
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
 
        // 設(shè)置最終的數(shù)據(jù)輸出的key和value 類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
 
        //設(shè)置數(shù)據(jù)的輸入路徑和輸出路徑
        FileInputFormat.setInputPaths(job, new Path("D:\\hadoop\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\hadoop\\output3"));
        //提交job
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
 
    }
 
}

最后運行  

出現(xiàn)了bug  經(jīng)過兩個小時的調(diào)試  找出答案   是在driver類中設(shè)置mapper類輸出kv類型出現(xiàn)差錯沒有配型成功 

現(xiàn)在是運作正確的

以上就是關(guān)于“Hadoop之Mapreduce序列化怎么實現(xiàn)”這篇文章的內(nèi)容，相信大家都有了一定的了解，希望小編分享的內(nèi)容對大家有幫助，若想了解更多相關(guān)的知識內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道。