2. 程式人生 > >hadoop 二次排序和一個java實現

hadoop 二次排序和一個java實現

阿新 • • 發佈：2018-12-16

需要二次排序的原因：mapreduce架構自動對對映器生成的鍵進行排序，即歸約器啟動之前，所有鍵是有序的，但是值是隨機的，二次排序指的是對值進行排序。歸約器輸入形如：(key, list(value)), list(value) = (V_{1}, V_{2},...,V_{n})，即一個key對應多個值，這些值是無序的，排序後得到有序的值，如下：

SORT(V_{1},V_{2},...,V_{n}) = (S_{1},S_{2},...,S_{n})

list(value) = (S_{1},S_{2},...,S_{n})

其中，S按照升序或者降序排列

歸約器對於二次排序的兩種解決方案：

1.讓歸約器讀取和快取給定鍵的所有值，完成歸約器中排序，特點是不可伸縮，依賴歸約器記憶體

2.使用mapreduce框架對歸約器值排序，方法是建立組合鍵，例如，A是鍵，B和C是值，選擇B作為次鍵，這樣A和B作為組合鍵，day作為值，將排序交給MapReduce框架完成，這樣不用在記憶體中排序，是可伸縮的方案

定製外掛：

1.分割槽器：根據對映器的輸出鍵決定將哪個對映器的輸出傳送到哪個歸約器，其本質是利用一致性hash演算法

2.比較器：按照自然鍵對一個歸約器中的資料分組，程式碼如下：

以year，month，temperature為例，MapReduce框架對於二次排序整體的處理流程是：

1.對映器建立(K,V)對，其中K是組合鍵(year,month,temperature)，V是temperature的值，組合鍵的(year,month)部分是自然鍵

2.通過分割槽器外掛，將所有自然鍵傳送給同一個歸約器

3.通過分組比較器，保證溫度按照順序到達歸約器

顯然，MapReduce框架完成了排序，而不用在記憶體中操作

程式碼如下，除了mapper，reducer，主作業流程以外，還有其餘3個檔案，一個分割槽器、一個比較器、一箇中間件類：

DateTemperatureGroupingComparator.java分組比較器：

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class DateTemperatureGroupingComparator extends WritableComparator {
    public DateTemperatureGroupingComparator() {
        /* 呼叫父類建構函式 */
        super(DateTemperaturePair.class, true);
    }
    @Override
    /* 決定輸出鍵和歸約器的對應關係，保證相同鍵傳送到同一個歸約器 */
    public int compare(WritableComparable a, WritableComparable b) {
        DateTemperaturePair pair1 = (DateTemperaturePair) a;
        DateTemperaturePair pair2 = (DateTemperaturePair) b;
        return pair1.getYearMonth().compareTo(pair2.getYearMonth());
    }
}

DateTemperaturePartitioner.java分割槽器：

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class DateTemperaturePartitioner extends
    Partitioner<DateTemperaturePair, Text> {
    @Override
    /* 入參是mapper的輸出鍵和輸出值的型別，是一個String類的內建hash演算法 */
    public int getPartition(DateTemperaturePair dataTemperaturePair, Text text,
        int i) {
        return Math.abs(dataTemperaturePair.getYearMonth().hashCode() % i);
    }
}

DateTemperaturePair.java中間鍵類

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

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

public class DateTemperaturePair
    /* java不支援多重繼承，使用implements繼承介面，不同介面之間用逗號隔開 */
    implements Writable, WritableComparable<DateTemperaturePair> {
    private String yearMonth;   //自然鍵
    private String day;
    protected Integer temperature;  //次鍵

    /* 用這個方法指出如何對DateTemperaturePair排序 */
    public int compareTo(DateTemperaturePair o) {
        /* 呼叫String的字串比較方法compareTo */
        int compareValue = this.yearMonth.compareTo(o.getYearMonth());
        if (compareValue == 0) {
            compareValue = temperature.compareTo(o.getTemperature());
        }
        /* 這樣實現降序排列 */
        return -1 * compareValue;
    }

    /* DataOutput用於將java基本型別轉換成二進位制字元流 */
    public void write(DataOutput dataOutput) throws IOException {
        Text.writeString(dataOutput, yearMonth);
        dataOutput.writeInt(temperature);
    }

    public void readFields(DataInput dataInput) throws IOException {
        this.yearMonth = Text.readString(dataInput);
        this.temperature = dataInput.readInt();
    }

    @Override
    public String toString() {
        return yearMonth.toString();
    }

    public String getYearMonth() {
        return yearMonth;
    }

    public void setYearMonth(String text) {
        this.yearMonth = text;
    }

    public void setDay(String day) {
        this.day = day;
    }

    public Integer getTemperature() {
        return temperature;
    }

    public void setTemperature(Integer temperature) {
        this.temperature = temperature;
    }
}

SecondarySortingMapper.java對映器：

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/* 注意輸出鍵的型別是DateTemperaturePair，是自定義的組合鍵 */
public class SecondarySortingMapper extends
        Mapper<LongWritable, Text, DateTemperaturePair, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
        String[] tokens = value.toString().split(",");
        String yearMonth = tokens[0] + "-" + tokens[1];
        String day = tokens[2];
        int temperature = Integer.parseInt(tokens[3]);
        DateTemperaturePair reduceKey = new DateTemperaturePair();
        reduceKey.setYearMonth(yearMonth);
        reduceKey.setDay(day);
        reduceKey.setTemperature(temperature);
        context.write(reduceKey, new IntWritable(temperature));
    }
}

SecondarySortingReducer.java歸約器：

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/* 輸入鍵和輸出鍵的型別都是自定義的中間鍵 */
public class SecondarySortingReducer extends
        Reducer<DateTemperaturePair, IntWritable, DateTemperaturePair, Text> {
    @Override
    protected void reduce(DateTemperaturePair key,
            Iterable<IntWritable> values, Context context) throws IOException,
            InterruptedException {
        /* java的字串變數，在修改場景下執行速度和效率比string高 */
        StringBuilder sortedTemperatureList = new StringBuilder();
        for (IntWritable temperature : values) {
            sortedTemperatureList.append(temperature);
            sortedTemperatureList.append(",");
        }
        sortedTemperatureList.deleteCharAt(sortedTemperatureList.length() - 1);
        context.write(key, new Text(sortedTemperatureList.toString()));
    }
}

SecondarySort.java主作業流程：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


public class SecondarySort extends Configured implements Tool {
    public int run(String[] args) throws Exception {

        /* 設定作業，指導hadoop獲取jar包 */
        Job job = new Job();
        job.setJarByClass(SecondarySort.class);
        job.setJobName("SecondarySort");

        /* 獲取input路徑和output路徑 */
        Path inputPath = new Path(args[0]);
        Path outputPath = new Path(args[1]);
        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        /* 設定mapper的輸出鍵和輸出值 */
        job.setMapOutputKeyClass(DateTemperaturePair.class);
        job.setMapOutputValueClass(IntWritable.class);
       
        /* 設定reducer的輸出鍵和輸出值 */
        job.setOutputKeyClass(DateTemperaturePair.class);
        job.setOutputValueClass(IntWritable.class);
       
        /* 指定要使用的mapper，reducer，分割槽器，比較器 */
        job.setMapperClass(SecondarySortingMapper.class);
        job.setReducerClass(SecondarySortingReducer.class);
        job.setPartitionerClass(DateTemperaturePartitioner.class);
        job.setGroupingComparatorClass(DateTemperatureGroupingComparator.class);

        boolean status = job.waitForCompletion(true);
        return status ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
             throw new IllegalArgumentException(
                "!!!!!!!!!!!!!! Usage!!!!!!!!!!!!!!: SecondarySort"
                + "<input-path> <output-path>");
        }
        int returnStatus = ToolRunner.run(new SecondarySort(), args);
        System.exit(returnStatus);
    }
}

輸入檔案：

[[email protected] ~]# hdfs dfs -cat /sample_input.txt
2000,12,04,10
2000,11,01,20
2000,12,02,-20
2000,11,02,30
2000,11,24,-40
2012,12,21,30
2012,12,22,-20
2012,12,23,60
2012,12,24,70
2012,12,25,10
2013,01,22,80
2013,01,23,90
2013,01,24,70
2013,01,20,-10

執行作業命令：

hadoop jar SecondarySort.jar SecondarySort /sample_input.txt output

作業執行結果如下，可以看到已經按照temperature欄位降序排序了：

[[email protected] ~]# hdfs dfs -cat output/*
2013-01	90,80,70,-10
2012-12	70,60,30,10,-20
2000-12	10,-20
2000-11	30,20,-40

相關推薦

hadoop 二次排序和一個java實現

需要二次排序的原因：mapreduce架構自動對對映器生成的鍵進行排序，即歸約器啟動之前，所有鍵是有序的，但是值是隨機的，二次排序指的是對值進行排序。歸約器輸入形如：，即一個key對應多個值，這些值是無序的，排序後得到有序的值，如下： 其中，S按照升序或者降序排列

hadoop二次排序的原理和實現

預設情況下，Map輸出的結果會對Key進行預設的排序，但是有時候需要對Key排序的同時還需要對Value進行排序，這時候就要用到二次排序了。下面我們來說說二次排序 1、二次排序原理   我們把二次排序分為以下幾個階段   Map起始階段     在Map階段，使用jo

一起學Hadoop——二次排序演算法的實現

二次排序，從字面上可以理解為在對key排序的基礎上對key所對應的值value排序，也叫輔助排序。一般情況下，MapReduce框架只對key排序，而不對key所對應的值排序，因此value的排序經常是不固定的。但是我們經常會遇到同時對key和value排序的需求，例如Hadoop權威指南中的求一年的高高氣溫

hadoop二次排序 (Map/Reduce中分割槽和分組的問題)

1.二次排序概念：首先按照第一欄位排序，然後再對第一欄位相同的行按照第二欄位排序，注意不能破壞第一次排序的結果 。如： 輸入檔案：20 21 50 51 50 52 50 53 50 54 60 51 60 53 60 52 60 56 60 57 70 58 60 61 70 54 70 55 70 56 

Spark 二次排序自定義key 實現(Java)

楔子 spark java版本的二次排序 實現 資料如下 2::4 2::10 3::6 1::5 按照 第一列 和第二列 倒敘排列 實現如下的結果 3::6 2::10 2::4 1::5 demo GitHub 位置 的 Second

Hadoop 二次排序實現

業務場景:通常情況下,在MR操作中到達Reduce中的key值都是按照指定的規則進行排序,在單一key的情況下一切都進行的很自然,直到我們要求資料不再單純的按key進行排序,以如下資料舉例: Key   ->      value: 100  ->      2

hadoop二次排序實現join

package join; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import java.util.Iterator; import org.apac

hadoop 二次排序的一些思考

先說一下mr的二次排序需求： 假如檔案有兩列分別為name、score，需求是先按照name排序，name相同按照score排序 資料如下： jx 20 gj 30 jx 10 gj 15 輸出結果要求： gj 15 gj 30 jx 10 jx 20 我們常見的實現思路是： 1. 自

二叉樹基礎操作 ，前中後序遍歷，求二叉樹高度，二叉搜尋樹（二叉排序樹）Java實現 程式碼集合

首先，定義一個樹類Tree.java public class Tree { public TreeNode root; } 定義樹節點類TreeNode.java public class TreeNode { public TreeNode(int

hadoop二次排序

趁這個時候，順便把hadoop的用於比較的Writable, WritableComparable, Comprator等搞清楚。。 1.二次排序概念： 首先按照第一欄位排序，然後再對第一欄位相同的行按照第二欄位排序，注意不能破壞第一次排序的結果 。 如： 輸入檔案： 20 21 50 51

Hadoop二次排序及MapReduce處理流程例項詳解

一、概述 MapReduce框架對處理結果的輸出會根據key值進行預設的排序，這個預設排序可以滿足一部分需求，但是也是十分有限的，在我們實際的需求當中，往往有要對reduce輸出結果進行二次排序的需求。對於二次排序的實現，網路上已經有很多人分享過了，但是對二次排序的實現原理

二叉排序樹，java實現（知識簡單的實現，持續完善更新）

定義： 二叉排序樹就是左子樹都比節點小，右子樹都比節點大。簡單的排序二叉樹實現。 程式碼： package com.wzq.data_structure; public class Bina

二叉排序樹的java實現

最近在看JDK原始碼時，發現JDK的J.U.C包下面的很多類都用到了二叉排序樹和紅黑樹，就想著這一塊還是考研的時候看了的，順便就在理論基礎上覆習加用Java做個實現。 二叉排序樹 先來說說這個二叉排序樹的定義和性質： 定義：二叉排序樹或者

spark學習記錄（七、二次排序和分組取TopN問題）

1.二次排序 例題：將兩列數字按第一列升序，如果第一列相同，則第二列升序排列 資料檔案：https://download.csdn.net/download/qq_33283652/10894807 將資料封裝成物件，對物件進行排序，然後取出value public class Se

Hadoop 二次排序 Secondary Sort

mr自帶的例子中的原始碼SecondarySort，我重新寫了一下，基本沒變。 這個例子中定義的map和reduce如下，關鍵是它對輸入輸出型別的定義：（java泛型程式設計）  public static class Map extends Mapp

二叉排序樹建立(JAVA實現)

      最近看了一下二叉排序樹的建立，自己寫了一段程式碼，用來建立二叉排序樹,給定一個數組，對這個陣列中的數字進行建立二叉排序樹。分兩種情況：      1  陣列中的數字是隨機的，也就是說沒有順序 eg : int  a [ ] = {3,1,2,5,0,7,9,8}

Python Hadoop Mapreduce 實現Hadoop Streaming分組和二次排序

需求：公司給到一份全國各門店銷售資料，要求：1.按門店市場分類，將同一市場的門店放到一起；2.將各家門店按銷售額從大到小，再按利潤從大到小排列 一 需求一：按市場對門店進行分組 分組(partition) Hadoop streaming框架預設情況下會以’/t

Hadoop和Spark分別實現二次排序

將下列資料中每個分割槽中的第一列順序排列，第二列倒序排列。 Text  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 2021 5051

分別使用Hadoop和Spark實現二次排序

零、序（注意本部分與標題無太大關係，可直接翻到第一部分） 　　既然沒用為啥會有序？原因不想再開一篇文章，來抒發點什麼感想或者計劃了，就在這裡寫點好了： 　　前些日子買了幾本書，打算學習和研究大資料方面的知識，一直因為實習、考試、畢業設計等問題搞得沒有時間，現在進入了寒

Hadoop MapReduce二次排序演算法與實現之演算法解析

MapReduce二次排序的原理     1.在Mapper階段，會通過inputFormat的getSplits來把資料集分割成split public abstract class Input