hadoop如何自定義分區

今天小編給大家分享一下hadoop如何自定義分區的相關知識點，內容詳細，邏輯清晰，相信大部分人都還太了解這方面的知識，所以分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后有所收獲，下面我們一起來了解一下吧。

分區概念

分區這個詞對很多同學來說并不陌生，比如Java很多中間件中，像kafka的分區，mysql的分區表等，分區存在的意義在于將數據按照業務規則進行合理的劃分，方便后續對各個分區數據高效處理

Hadoop分區

hadoop中的分區，是把不同數據輸出到不同reduceTask ，最終到輸出不同文件中

hadoop 默認分區規則

• hash分區

• 按照key的hashCode % reduceTask 數量 = 分區號

• 默認reduceTask 數量為1，當然也可以在driver 端設置

以下是Partition 類中摘取出來的源碼，還是很容易懂的

hash分區代碼演示

下面是wordcount案例中的driver部分的代碼，默認情況下我們不做任何設置，最終輸出一個統計單詞個數的txt文件，如果我們在這段代碼中添加這樣一行

再次運行下面的程序后，會出現什么結果呢？

public class DemoJobDriver {

    public static void main(String[] args) throws Exception {

        //1、獲取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、設置jar路徑
        job.setJarByClass(DemoJobDriver.class);

        //3、關聯mapper 和 Reducer
        job.setMapperClass(DemoMapper.class);
        job.setReducerClass(DemoReducer.class);

        //4、設置 map輸出的 key/val 的類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //5、設置最終輸出的key / val 類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //6、設置最終的輸出路徑
        String inputPath = "F:\\網盤\\csv\\hello.txt";
        String outPath = "F:\\網盤\\csv\\wordcount\\hello_result.txt";

        //設置輸出文件為2個
        job.setNumReduceTasks(2);

        FileInputFormat.setInputPaths(job,new Path(inputPath));
        FileOutputFormat.setOutputPath(job,new Path(outPath));

        // 7 提交job
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

可以看到，最終輸出了2個統計結果文件，每個文件中的內容有所不同，這就是默認情況下，當reducer個數設置為多個時，會按照hash分區算法計算結果并輸出到不同分區對應的文件中去

自定義分區步驟

• 自定義類繼承Partitioner

• 重寫getPartition方法，并在此方法中根據業務規則控制不同的數據進入到不同分區

• 在Job的驅動類中，設置自定義的Partitioner類

• 自定義Partition后，要根據自定義的Partition邏輯設置相應數量的ReduceTask

業務需求

將下面文件中 的人物名稱按照姓氏，“馬”姓的放入第一個分區，“李”姓的放入第二個分區，其他的放到其他第三個分區中

自定義分區

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.io.Text;

public class MyPartioner extends Partitioner<Text, IntWritable> {

    @Override
    public int getPartition(Text text, IntWritable intWritable, int partion) {
        String key = text.toString();
        if(StringUtils.isNotEmpty(key.trim())){
            if(key.startsWith("馬")){
                partion = 0;
            }else if(key.startsWith("李")){
                partion = 1;
            }else {
                partion = 2;
            }
        }
        return partion;
    }
}

將自定義分區關聯到Driver類中，注意這里的ReduceTasks個數和自定義的分區數量保持一致

job.setNumReduceTasks(3);
job.setPartitionerClass(MyPartioner.class);

下面運行Driver類，觀察最終的輸出結果，也是按照預期，將不同的姓氏數據輸出到了不同的文件中

關于自定義分區的總結

• 如果ReduceTask的數量 > 自定義partion中的分區數量，則會多產生幾個空的輸出文件

• 如果 1 < ReduceTask < 自定義partion中的分區數量，有一部分的數據處理過程中無法找到相應的分區文件存儲，會拋異常

• 如果ReduceTask = 1 ，則不管自定義的partion中分區數量為多少個，最終結果都只會交給這一個ReduceTask 處理，最終只會產生一個結果文件

• 分區號必須從0開始，逐一累加

以上就是“hadoop如何自定義分區”這篇文章的所有內容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學習更多的知識，請關注億速云行業資訊頻道。