getPartition中的引數numPartitions是怎么獲取的？

 public int getPartition(IntWritable key, IntWritable value, int numPartitions)

import org.apache.hadoop.conf.Configuration;

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.Mapper;

import org.apache.hadoop.mapreduce.Partitioner;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.util.GenericOptionsParser;



import java.io.IOException;



/**

 * Created by dell on 2017/9/25.

 * @auther w

 * 

 */

public class MySort {



    static final String INPUT_PATH = "hdfs://hadoopwang0:9000/test";

    static final String OUT_PATH = "hdfs://hadoopwang0:9000/testout";

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {





        Configuration conf = new Configuration();

//        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

//        if (otherArgs.length != 2) {

//            System.err.println("Usage: wordcount <in> <out>");

//            System.exit(2);

//        }



        Job job = new Job(conf, "MySort");

        job.setJarByClass(MySort.class);

        job.setMapperClass(MyMap.class);

        job.setReducerClass(MyReduce.class);

        job.setPartitionerClass(MyPartition.class);

        job.setOutputKeyClass(IntWritable.class);

        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job, new Path(INPUT_PATH));

        FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));

        System.exit(job.waitForCompletion(true) ? 0:1);

    }



    //Map方法：將輸入的value轉化為IntWritable型別，作為輸出的Key。

    public static class MyMap extends Mapper<Object, Text, IntWritable, IntWritable>{

        private static  IntWritable data = new IntWritable();



        @Override

        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();



            data.set(Integer.parseInt(line));

            context.write(data, new IntWritable(1));

        }



    }



    //Reduce方法：將輸入的Key復制到輸出的value中，然后根據輸入的<value-list>中元素的個數決定Key的輸出次數

    //全域用linenum來代表key的位次

    public  static class MyReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable >{

        private static IntWritable linenum = new IntWritable(1);



        @Override

        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

            System.out.println("Reducer:"+key);

            for (IntWritable val : values) {

                context.write(linenum, key);

                linenum = new IntWritable(linenum.get() + 1);

            }

        }

    }

    //自定義Partition函式:此函式根據輸入的資料的最大值和MapReduce框架中的partition數量獲取將輸入資料按照

    //按照大小分塊的邊界，然后根據輸入數值和邊界關系回傳對應的Partiton ID

    public static class MyPartition extends Partitioner<IntWritable, IntWritable>{

        public int getPartition(IntWritable key, IntWritable value, int numPartitions) {

            int Maxnumber = 6522;

            int bound = Maxnumber / numPartitions + 1;

            int Keynumber = key.get();

            for (int i = 0; i < numPartitions; i++) {

                if (Keynumber < bound * i && Keynumber >= bound * (i - 1)) {

                    return i - 1;

                }

            }

            return -1;

        }

    }





}

uj5u.com熱心網友回復：

uj5u.com熱心網友回復：

標籤：Spark

上一篇：浪潮的服務器有個紅閃電燈亮了

下一篇：aws 上的rds sqlserver 服務怎么開啟發布（Replication）功能？