自定義 Partitioner

主要的邏輯就在這里, 這也是這個案例的意義, 通過 Partitioner 將數(shù)據(jù)分發(fā)給不同的Reducer

/** * 這里的輸入類型與我們map階段的輸出類型相同 */ public class MyPartitioner extends Partitioner<Text,NullWritable>{/*** 返回值表示我們的數(shù)據(jù)要去到哪個分區(qū)* 返回值只是一個分區(qū)的標記，標記所有相同的數(shù)據(jù)去到指定的分區(qū)*/@Overridepublic int getPartition(Text text, NullWritable nullWritable, int i){String result = text.toString().split("\t")[5]; System.out.println(result);if (Integer.parseInt(result) > 15){return 1;}else{return 0;}} }

Main 入口

public class PartitionMain extends Configured implements Tool {public static void main(String[] args) throws Exception{int run = ToolRunner.run(new Configuration(), newPartitionMain(), args);System.exit(run);}@Overridepublic int run(String[] args) throws Exception {Job job = Job.getInstance(super.getConf(),PartitionMain.class.getSimpleName());job.setJarByClass(PartitionMain.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);TextInputFormat.addInputPath(job,newPath("hdfs://192.168.52.250:8020/partitioner"));TextOutputFormat.setOutputPath(job,newPath("hdfs://192.168.52.250:8020/outpartition"));job.setMapperClass(MyMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);job.setReducerClass(MyReducer.class);/*** 設置我們的分區(qū)類，以及我們的reducetask的個數(shù)，注意reduceTask的個數(shù)一定要與我們的* 分區(qū)數(shù)保持一致*/job.setPartitionerClass(MyPartitioner.class);job.setNumReduceTasks(2);boolean b = job.waitForCompletion(true);return b?0:1;}}

?

超強干貨來襲 云風專訪：近40年碼齡，通宵達旦的技術人生

總結(jié)

以上是生活随笔為你收集整理的MapReduce分区-代码实现的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。