MapReduce 排序和序列化代碼實現(xiàn)&規(guī)約

MapReduce 排序和序列化

序列化 (Serialization) 是指把結(jié)構(gòu)化對象轉(zhuǎn)化為字節(jié)流
反序列化 (Deserialization) 是序列化的逆過程. 把字節(jié)流轉(zhuǎn)為結(jié)構(gòu)化對象. 當(dāng)要在進程間傳遞對象或持久化對象的時候, 就需要序列化對象成字節(jié)流, 反之當(dāng)要將接收到或從磁盤讀取的字節(jié)流轉(zhuǎn)換為對象, 就要進行反序列化
Java 的序列化 (Serializable) 是一個重量級序列化框架, 一個對象被序列化后, 會附帶很多額外的信息 (各種校驗信息, header, 繼承體系等）, 不便于在網(wǎng)絡(luò)中高效傳輸. 所以, Hadoop 自己開發(fā)了一套序列化機制(Writable), 精簡高效. 不用像 Java 對象類一樣傳輸多層的父子關(guān)系, 需要哪個屬性就傳輸哪個屬性值, 大大的減少網(wǎng)絡(luò)傳輸?shù)拈_銷
Writable 是 Hadoop 的序列化格式, Hadoop 定義了這樣一個 Writable 接口. 一個類要支持可序列化只需實現(xiàn)這個接口即可
另外 Writable 有一個子接口是 WritableComparable, WritableComparable 是既可實現(xiàn)序列化, 也可以對key進行比較, 我們這里可以通過自定義 Key 實現(xiàn) WritableComparable 來實現(xiàn)我們的排序功能

數(shù)據(jù)格式如下

要求:

第一列按照字典順序進行排列
第一列相同的時候, 第二列按照升序進行排列

解決思路:

將 Map 端輸出的 <key,value> 中的 key 和 value 組合成一個新的 key (newKey), value值不變
這里就變成 <(key,value),value>, 在針對 newKey 排序的時候, 如果 key 相同, 就再對value進行排序

Step 1. 自定義類型和比較器

public class SortBean implements WritableComparable<SortBean>{

    private String word;
    private int  num;

    public String getWord() {
        return word;
    }

    public void setWord(String word) {
        this.word = word;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }

    @Override
    public String toString() {
        return   word + "\t"+ num ;
    }

    //實現(xiàn)比較器乙濒，指定排序的規(guī)則
    /*
      規(guī)則:
        第一列(word)按照字典順序進行排列    //  aac   aad
        第一列相同的時候, 第二列(num)按照升序進行排列
     */
    /*
        a  1
        a  5
        b  3
        b  8
     */
    @Override
    public int compareTo(SortBean sortBean) {
        //先對第一列排序: Word排序
        int result = this.word.compareTo(sortBean.word);
        //如果第一列相同详羡，則按照第二列進行排序
        if(result == 0){
            return  this.num - sortBean.num;
        }
        return result;
    }

    //實現(xiàn)序列化
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(word);
        out.writeInt(num);
    }

    //實現(xiàn)反序列
    @Override
    public void readFields(DataInput in) throws IOException {
            this.word = in.readUTF();
            this.num = in.readInt();
    }
}

Step 2. Mapper

public class SortMapper extends Mapper<LongWritable,Text,SortBean,NullWritable> {
    /*
      map方法將K1和V1轉(zhuǎn)為K2和V2:

      K1            V1
      0            a  3
      5            b  7
      ----------------------
      K2                         V2
      SortBean(a  3)         NullWritable
      SortBean(b  7)         NullWritable
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:將行文本數(shù)據(jù)(V1)拆分，并將數(shù)據(jù)封裝到SortBean對象,就可以得到K2
        String[] split = value.toString().split("\t");

        SortBean sortBean = new SortBean();
        sortBean.setWord(split[0]);
        sortBean.setNum(Integer.parseInt(split[1]));

        //2:將K2和V2寫入上下文中
        context.write(sortBean, NullWritable.get());
    }
}

Step 3. Reducer

public class SortReducer extends Reducer<SortBean,NullWritable,SortBean,NullWritable> {

    //reduce方法將新的K2和V2轉(zhuǎn)為K3和V3
    @Override
    protected void reduce(SortBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
       context.write(key, NullWritable.get());
    }
}

Step 4. Main 入口

public class JobMain extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception {
        //1:創(chuàng)建job對象
        Job job = Job.getInstance(super.getConf(), "mapreduce_sort");

        //2:配置job任務(wù)(八個步驟)
            //第一步:設(shè)置輸入類和輸入的路徑
            job.setInputFormatClass(TextInputFormat.class);
            ///TextInputFormat.addInputPath(job, new Path("hdfs://node01:8020/input/sort_input"));
            TextInputFormat.addInputPath(job, new Path("file:///D:\\input\\sort_input"));

            //第二步: 設(shè)置Mapper類和數(shù)據(jù)類型
            job.setMapperClass(SortMapper.class);
            job.setMapOutputKeyClass(SortBean.class);
            job.setMapOutputValueClass(NullWritable.class);

            //第三邻吞，四齿兔，五橱脸，六

            //第七步：設(shè)置Reducer類和類型
            job.setReducerClass(SortReducer.class);
            job.setOutputKeyClass(SortBean.class);
            job.setOutputValueClass(NullWritable.class);


            //第八步: 設(shè)置輸出類和輸出的路徑
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\sort_out"));


        //3:等待任務(wù)結(jié)束
        boolean bl = job.waitForCompletion(true);

        return bl?0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        //啟動job任務(wù)
        int run = ToolRunner.run(configuration, new JobMain(), args);

        System.exit(run);
    }
}

規(guī)約Combiner

概念

每一個 map 都可能會產(chǎn)生大量的本地輸出，Combiner 的作用就是對 map 端的輸出先做一次合并分苇，以減少在 map 和 reduce 節(jié)點之間的數(shù)據(jù)傳輸量添诉，以提高網(wǎng)絡(luò)IO 性能，是 MapReduce 的一種優(yōu)化手段之一

combiner 是 MR 程序中 Mapper 和 Reducer 之外的一種組件
combiner 組件的父類就是 Reducer
combiner 和 reducer 的區(qū)別在于運行的位置
- Combiner 是在每一個 maptask 所在的節(jié)點運行
- Reducer 是接收全局所有 Mapper 的輸出結(jié)果
combiner 的意義就是對每一個 maptask 的輸出進行局部匯總医寿，以減小網(wǎng)絡(luò)傳輸量

實現(xiàn)步驟

自定義一個 combiner 繼承 Reducer栏赴，重寫 reduce 方法
在 job 中設(shè)置 job.setCombinerClass(CustomCombiner.class)

combiner 能夠應(yīng)用的前提是不能影響最終的業(yè)務(wù)邏輯，而且靖秩，combiner 的輸出 kv 應(yīng)該跟 reducer 的輸入 kv 類型要對應(yīng)起來

MapReduce案例-流量統(tǒng)計

需求一: 統(tǒng)計求和

統(tǒng)計每個手機號的上行數(shù)據(jù)包總和须眷，下行數(shù)據(jù)包總和，上行總流量之和沟突，下行總流量之和
分析：以手機號碼作為key值花颗，上行流量，下行流量惠拭，上行總流量扩劝，下行總流量四個字段作為value值，然后以這個key职辅，和value作為map階段的輸出棒呛，reduce階段的輸入

Step 1: 自定義map的輸出value對象FlowBean

public class FlowBean implements Writable {
    private Integer upFlow;  //上行數(shù)據(jù)包數(shù)
    private Integer downFlow;  //下行數(shù)據(jù)包數(shù)
    private Integer upCountFlow; //上行流量總和
    private Integer downCountFlow;//下行流量總和

    public Integer getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(Integer upFlow) {
        this.upFlow = upFlow;
    }

    public Integer getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(Integer downFlow) {
        this.downFlow = downFlow;
    }

    public Integer getUpCountFlow() {
        return upCountFlow;
    }

    public void setUpCountFlow(Integer upCountFlow) {
        this.upCountFlow = upCountFlow;
    }

    public Integer getDownCountFlow() {
        return downCountFlow;
    }

    public void setDownCountFlow(Integer downCountFlow) {
        this.downCountFlow = downCountFlow;
    }

    @Override
    public String toString() {
        return  upFlow +
                "\t" + downFlow +
                "\t" + upCountFlow +
                "\t" + downCountFlow;
    }

    //序列化方法
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(upFlow);
        out.writeInt(downFlow);
        out.writeInt(upCountFlow);
        out.writeInt(downCountFlow);
    }

    //反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readInt();
        this.downFlow = in.readInt();
        this.upCountFlow = in.readInt();
        this.downCountFlow = in.readInt();
    }
}

Step 2: 定義FlowMapper類

public class FlowCountMapper extends Mapper<LongWritable,Text,Text,FlowBean> {
    /*
      將K1和V1轉(zhuǎn)為K2和V2:
      K1              V1
      0            1360021750219    128 1177    16852   200
     ------------------------------
      K2              V2
      13600217502     FlowBean(19   128 1177    16852)
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:拆分行文本數(shù)據(jù),得到手機號--->K2
        String[] split = value.toString().split("\t");
        String phoneNum = split[1];

        //2:創(chuàng)建FlowBean對象,并從行文本數(shù)據(jù)拆分出流量的四個四段,并將四個流量字段的值賦給FlowBean對象
        FlowBean flowBean = new FlowBean();

        flowBean.setUpFlow(Integer.parseInt(split[6]));
        flowBean.setDownFlow(Integer.parseInt(split[7]));
        flowBean.setUpCountFlow(Integer.parseInt(split[8]));
        flowBean.setDownCountFlow(Integer.parseInt(split[9]));

        //3:將K2和V2寫入上下文中
        context.write(new Text(phoneNum), flowBean);

    }
}

Step 3: 定義FlowReducer類

public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> {
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        //1:遍歷集合,并將集合中的對應(yīng)的四個字段累計
         Integer upFlow = 0;  //上行數(shù)據(jù)包數(shù)
         Integer downFlow = 0;  //下行數(shù)據(jù)包數(shù)
         Integer upCountFlow = 0; //上行流量總和
         Integer downCountFlow = 0;//下行流量總和

        for (FlowBean value : values) {
            upFlow += value.getUpFlow();
            downFlow += value.getDownFlow();
            upCountFlow += value.getUpCountFlow();
            downCountFlow += value.getDownCountFlow();
        }

        //2:創(chuàng)建FlowBean對象,并給對象賦值  V3
        FlowBean flowBean = new FlowBean();
        flowBean.setUpFlow(upFlow);
        flowBean.setDownFlow(downFlow);
        flowBean.setUpCountFlow(upCountFlow);
        flowBean.setDownCountFlow(downCountFlow);

        //3:將K3和V3下入上下文中
        context.write(key, flowBean);
    }
}

Step 4: 程序main函數(shù)入口FlowMain

public class JobMain extends Configured implements Tool {

    //該方法用于指定一個job任務(wù)
    @Override
        public int run(String[] args) throws Exception {
        //1:創(chuàng)建一個job任務(wù)對象
        Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount");
        //如果打包運行出錯，則需要加該配置
        job.setJarByClass(JobMain.class);
        //2:配置job任務(wù)對象(八個步驟)

        //第一步:指定文件的讀取方式和讀取路徑
        job.setInputFormatClass(TextInputFormat.class);
        //TextInputFormat.addInputPath(job, new Path("hdfs://node01:8020/wordcount"));
        TextInputFormat.addInputPath(job, new Path("file:///D:\\input\\flowcount_input"));



        //第二步:指定Map階段的處理方式和數(shù)據(jù)類型
         job.setMapperClass(FlowCountMapper.class);
         //設(shè)置Map階段K2的類型
          job.setMapOutputKeyClass(Text.class);
        //設(shè)置Map階段V2的類型
          job.setMapOutputValueClass(FlowBean.class);


          //第三（分區(qū)）域携，四 （排序）
          //第五步: 規(guī)約(Combiner)
          //第六步 分組


          //第七步：指定Reduce階段的處理方式和數(shù)據(jù)類型
          job.setReducerClass(FlowCountReducer.class);
          //設(shè)置K3的類型
           job.setOutputKeyClass(Text.class);
          //設(shè)置V3的類型
           job.setOutputValueClass(FlowBean.class);

           //第八步: 設(shè)置輸出類型
           job.setOutputFormatClass(TextOutputFormat.class);
           //設(shè)置輸出的路徑
           TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\flowcount_out"));



        //等待任務(wù)結(jié)束
           boolean bl = job.waitForCompletion(true);

           return bl ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        //啟動job任務(wù)
        int run = ToolRunner.run(configuration, new JobMain(), args);
        System.exit(run);

    }
}

需求二: 上行流量倒序排序（遞減排序）

分析簇秒，以需求一的輸出數(shù)據(jù)作為排序的輸入數(shù)據(jù)，自定義FlowBean,以FlowBean為map輸出的key涵亏，以手機號作為Map輸出的value宰睡，因為MapReduce程序會對Map階段輸出的key進行排序

Step 1: 定義FlowBean實現(xiàn)WritableComparable實現(xiàn)比較排序

Java 的 compareTo 方法說明:

compareTo 方法用于將當(dāng)前對象與方法的參數(shù)進行比較。
如果指定的數(shù)與參數(shù)相等返回 0气筋。
如果指定的數(shù)小于參數(shù)返回 -1拆内。
如果指定的數(shù)大于參數(shù)返回 1。

例如：o1.compareTo(o2); 返回正數(shù)的話宠默，當(dāng)前對象（調(diào)用 compareTo 方法的對象 o1）要排在比較對象（compareTo 傳參對象 o2）后面麸恍，返回負數(shù)的話，放在前面

public class FlowBean implements WritableComparable<FlowBean> {
    private Integer upFlow;  //上行數(shù)據(jù)包數(shù)
    private Integer downFlow;  //下行數(shù)據(jù)包數(shù)
    private Integer upCountFlow; //上行流量總和
    private Integer downCountFlow;//下行流量總和

    public Integer getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(Integer upFlow) {
        this.upFlow = upFlow;
    }

    public Integer getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(Integer downFlow) {
        this.downFlow = downFlow;
    }

    public Integer getUpCountFlow() {
        return upCountFlow;
    }

    public void setUpCountFlow(Integer upCountFlow) {
        this.upCountFlow = upCountFlow;
    }

    public Integer getDownCountFlow() {
        return downCountFlow;
    }

    public void setDownCountFlow(Integer downCountFlow) {
        this.downCountFlow = downCountFlow;
    }

    @Override
    public String toString() {
        return  upFlow +
                "\t" + downFlow +
                "\t" + upCountFlow +
                "\t" + downCountFlow;
    }

    //序列化方法
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(upFlow);
        out.writeInt(downFlow);
        out.writeInt(upCountFlow);
        out.writeInt(downCountFlow);
    }

    //反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readInt();
        this.downFlow = in.readInt();
        this.upCountFlow = in.readInt();
        this.downCountFlow = in.readInt();
    }

    //指定排序的規(guī)則
    @Override
    public int compareTo(FlowBean flowBean) {
       // return this.upFlow.compareTo(flowBean.getUpFlow()) * -1;
       return  flowBean.upFlow - this.upFlow ;
    }
}

Step 2: 定義FlowMapper

public class FlowSortMapper extends Mapper<LongWritable,Text,FlowBean,Text> {
    //map方法:將K1和V1轉(zhuǎn)為K2和V2
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:拆分行文本數(shù)據(jù)(V1),得到四個流量字段,并封裝FlowBean對象---->K2
        String[] split = value.toString().split("\t");

        FlowBean flowBean = new FlowBean();

        flowBean.setUpFlow(Integer.parseInt(split[1]));
        flowBean.setDownFlow(Integer.parseInt(split[2]));
        flowBean.setUpCountFlow(Integer.parseInt(split[3]));
        flowBean.setDownCountFlow(Integer.parseInt(split[4]));

        //2:通過行文本數(shù)據(jù),得到手機號--->V2
        String phoneNum = split[0];

        //3:將K2和V2下入上下文中
        context.write(flowBean, new Text(phoneNum));

    }
}

Step 3: 定義FlowReducer

/*
  K2: FlowBean
  V2: Text  手機號

  K3: Text  手機號
  V3: FlowBean
 */

public class FlowSortReducer extends Reducer<FlowBean,Text,Text,FlowBean> {
    @Override
    protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //1:遍歷集合,取出 K3,并將K3和V3寫入上下文中
        for (Text value : values) {
            context.write(value, key);
        }

    }
}

Step 4: 程序main函數(shù)入口

public class JobMain extends Configured implements Tool {

    //該方法用于指定一個job任務(wù)
    @Override
        public int run(String[] args) throws Exception {
        //1:創(chuàng)建一個job任務(wù)對象
        Job job = Job.getInstance(super.getConf(), "mapreduce_flowsort");


        //2:配置job任務(wù)對象(八個步驟)

        //第一步:指定文件的讀取方式和讀取路徑
        job.setInputFormatClass(TextInputFormat.class);
        //TextInputFormat.addInputPath(job, new Path("hdfs://node01:8020/wordcount"));
        TextInputFormat.addInputPath(job, new Path("file:///D:\\out\\flowcount_out"));



        //第二步:指定Map階段的處理方式和數(shù)據(jù)類型
         job.setMapperClass(FlowSortMapper.class);
         //設(shè)置Map階段K2的類型
          job.setMapOutputKeyClass(FlowBean.class);
        //設(shè)置Map階段V2的類型
          job.setMapOutputValueClass(Text.class);


          //第三（分區(qū)）搀矫，四 （排序）
          //第五步: 規(guī)約(Combiner)
          //第六步 分組


          //第七步：指定Reduce階段的處理方式和數(shù)據(jù)類型
          job.setReducerClass(FlowSortReducer.class);
          //設(shè)置K3的類型
           job.setOutputKeyClass(Text.class);
          //設(shè)置V3的類型
           job.setOutputValueClass(FlowBean.class);

           //第八步: 設(shè)置輸出類型
           job.setOutputFormatClass(TextOutputFormat.class);
           //設(shè)置輸出的路徑
           TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\flowsort_out"));



        //等待任務(wù)結(jié)束
           boolean bl = job.waitForCompletion(true);

           return bl ? 0:1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        //啟動job任務(wù)
        int run = ToolRunner.run(configuration, new JobMain(), args);
        System.exit(run);

    }
}

需求三: 手機號碼分區(qū)

在需求一的基礎(chǔ)上抹沪，繼續(xù)完善，將不同的手機號分到不同的數(shù)據(jù)文件的當(dāng)中去瓤球，需要自定義分區(qū)來實現(xiàn)融欧，這里我們自定義來模擬分區(qū)，將以下數(shù)字開頭的手機號進行分開

135 開頭數(shù)據(jù)到一個分區(qū)文件
136 開頭數(shù)據(jù)到一個分區(qū)文件
137 開頭數(shù)據(jù)到一個分區(qū)文件
其他分區(qū)

自定義分區(qū)

public class FlowCountPartition extends Partitioner<Text,FlowBean> {

    /*
      該方法用來指定分區(qū)的規(guī)則:
        135 開頭數(shù)據(jù)到一個分區(qū)文件
        136 開頭數(shù)據(jù)到一個分區(qū)文件
        137 開頭數(shù)據(jù)到一個分區(qū)文件
        其他分區(qū)

       參數(shù):
         text : K2   手機號
         flowBean: V2
         i   : ReduceTask的個數(shù)
     */
    @Override
    public int getPartition(Text text, FlowBean flowBean, int i) {
        //1:獲取手機號
        String phoneNum = text.toString();

        //2:判斷手機號以什么開頭,返回對應(yīng)的分區(qū)編號(0-3)
        if(phoneNum.startsWith("135")){
            return  0;
        }else  if(phoneNum.startsWith("136")){
            return  1;
        }else  if(phoneNum.startsWith("137")){
            return  2;
        }else{
            return 3;
        }

    }
}

作業(yè)運行設(shè)置

job.setPartitionerClass(FlowPartition.class);
job.setNumReduceTasks(4);

修改輸入輸出路徑, 并運行

TextInputFormat.addInputPath(job, new Path("file:///D:\\input\\flowpartition_input"));
TextOutputFormat.setOutputPath(job, new Path("file:///D:\\out\\flowpartiton_out"));

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者

人面猴
序言：七十年代末卦羡，一起剝皮案震驚了整個濱河市噪馏，隨后出現(xiàn)的幾起案子，更是在濱河造成了極大的恐慌绿饵，老刑警劉巖欠肾，帶你破解...
沈念sama閱讀 211,639評論 6贊 492
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場離奇詭異拟赊，居然都是意外死亡刺桃，警方通過查閱死者的電腦和手機，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 90,277評論 3贊 385
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進店門吸祟，熙熙樓的掌柜王于貴愁眉苦臉地迎上來瑟慈，“玉大人，你說我怎么就攤上這事屋匕》夂溃” “怎么了？”我有些...
開封第一講書人閱讀 157,221評論 0贊 348
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵炒瘟，是天一觀的道長吹埠。經(jīng)常有香客問我，道長疮装，這世上最難降的妖魔是什么缘琅？我笑而不...
開封第一講書人閱讀 56,474評論 1贊 283
?港島之戀（遺憾婚禮）
正文為了忘掉前任，我火速辦了婚禮廓推，結(jié)果婚禮上刷袍，老公的妹妹穿的比我還像新娘。我一直安慰自己樊展，他們只是感情好呻纹，可當(dāng)我...
茶點故事閱讀 65,570評論 6贊 386
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布堆生。她就那樣靜靜地躺著，像睡著了一般雷酪。火紅的嫁衣襯著肌膚如雪淑仆。梳的紋絲不亂的頭發(fā)上，一...
開封第一講書人閱讀 49,816評論 1贊 290
城市分裂傳說
那天哥力，我揣著相機與錄音蔗怠，去河邊找鬼。笑死吩跋，一個胖子當(dāng)著我的面吹牛寞射，可吹牛的內(nèi)容都是我干的。我是一名探鬼主播锌钮，決...
沈念sama閱讀 38,957評論 3贊 408
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼桥温，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了梁丘？” 一聲冷哼從身側(cè)響起策治，我...
開封第一講書人閱讀 37,718評論 0贊 266
萬榮殺人案實錄
序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎兰吟，沒想到半個月后通惫，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
沈念sama閱讀 44,176評論 1贊 303
?護林員之死
正文獨居荒郊野嶺守林人離奇死亡混蔼，尸身上長有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點故事閱讀 36,511評論 2贊 327
?白月光啟示錄
正文我和宋清朗相戀三年履腋，在試婚紗的時候發(fā)現(xiàn)自己被綠了。大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片惭嚣。...
茶點故事閱讀 38,646評論 1贊 340
活死人
序言：一個原本活蹦亂跳的男人離奇死亡遵湖，死狀恐怖，靈堂內(nèi)的尸體忽然破棺而出晚吞，到底是詐尸還是另有隱情延旧，我是刑警寧澤，帶...
沈念sama閱讀 34,322評論 4贊 330
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布槽地，位于F島的核電站迁沫，受9級特大地震影響，放射性物質(zhì)發(fā)生泄漏捌蚊。R本人自食惡果不足惜集畅，卻給世界環(huán)境...
茶點故事閱讀 39,934評論 3贊 313
男人毒藥：我在死后第九天來索命
文/蒙蒙一、第九天我趴在偏房一處隱蔽的房頂上張望缅糟。院中可真熱鬧挺智，春花似錦、人聲如沸窗宦。這莊子的主人今日做“春日...
開封第一講書人閱讀 30,755評論 0贊 21
一樁弒父案二鳄，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽。三九已至媒怯，卻和暖如春订讼，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背沪摄。一陣腳步聲響...
開封第一講書人閱讀 31,987評論 1贊 266
情欲美人皮
我被黑心中介騙來泰國打工，沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留纱烘，地道東北人杨拐。一個月前我還...
沈念sama閱讀 46,358評論 2贊 360
代替公主和親
正文我出身青樓，卻偏偏與公主長得像擂啥，于是被迫代替她去往敵國和親哄陶。傳聞我的和親對象是個殘疾皇子，可洞房花燭夜當(dāng)晚...
茶點故事閱讀 43,514評論 2贊 348

MapReduce 排序和序列化代碼實現(xiàn)&規(guī)約