hadoop-MapReduce分区

在 MapReduce 中, 通过我们指定分区, 会将同一个分区的数据发送到同一个 Reduce 当中进行处理

例如: 为了数据的统计, 可以把一批类似的数据发送到同一个 Reduce 当中, 在同一个 Reduce 当中统计相同类型的数据, 就可以实现类似的数据分区和统计等 其实就是相同类型的数据, 有共性的数据, 送到一起去处理

Reduce 当中默认的分区只有一个

需求：数据进行分开处理，将15以上的结果以及15以下的结果进行分开成两个文件进行保存。

Step 1. 定义 Mapper

这个 Mapper 程序不做任何逻辑, 也不对 Key-Value 做任何改变, 只是接收数据, 然后往下发送

public class MyMapper extends Mapper<LongWritable,Text,Text,NullWritable>{ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value,NullWritable.get()); } }

Step 2. 自定义 Partitioner

主要的逻辑就在这里, 这也是这个案例的意义, 通过 Partitioner 将数据分发给不同的 Reducer

/** * 这里的输入类型与我们map阶段的输出类型相同 */ public class MyPartitioner extends Partitioner<Text,NullWritable>{ /** * 返回值表示我们的数据要去到哪个分区 * 返回值只是一个分区的标记，标记所有相同的数据去到指定的分区 */ @Override public int getPartition(Text text, NullWritable nullWritable, int i) { String result = text.toString().split("\t")[5]; if (Integer.parseInt(result) > 15){ return 1; }else{ return 0; } } }

Step 3. 定义 Reducer 逻辑

这个 Reducer 也不做任何处理, 将数据原封不动的输出即可

public class MyReducer extends Reducer<Text,NullWritable,Text,NullWritable> { @Override protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { context.write(key,NullWritable.get()); } }

Step 4. 主类中设置分区类和ReduceTask个数

public class PartitionMain extends Configured implements Tool { public static void main(String[] args) throws Exception{ int run = ToolRunner.run(new Configuration(), new PartitionMain(), args); System.exit(run); } @Override public 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,new Path("hdfs://192.168.52.250:8020/partitioner")); TextOutputFormat.setOutputPath(job,new Path("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); /** * 设置我们的分区类，以及我们的reducetask的个数，注意reduceTask的个数一定要与我们的 * 分区数保持一致 */ job.setPartitionerClass(MyPartitioner.class); job.setNumReduceTasks(2); boolean b = job.waitForCompletion(true); return b?0:1; } }