Java大数据处理的核心技术MapReduce框架

mapreduce.input.fileinputformat.split.minsize=1 默认值为1 mapreduce.input.fileinputformat.split.maxsize=Long.MAXValue 默认值Long.MAXValue

public class ProvincePartitioner extends Partitioner { @Override public int getPartition(Text text, FlowBean flowBean, int i) { // 获取手机号前三位prePhone String phone = text.toString(); String prePhone = phone.substring(0, 3); //定义一个分区号变量partition,根据prePhone 设置分区号 int partition; if ("136".equals(prePhone)) { partition = 0; } else if ("137".equals(prePhone)) { partition = 1; } else if ("138".equals(prePhone)) { partition = 2; } else if ("139".equals(prePhone)) { partition = 3; } else { partition = 4; } //最后返回分区号partition return partition; } }

public class FlowDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1.获取job对象 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2.关联本Driver类 job.setJarByClass(FlowDriver.class); // 3.关联Mapper和Reducer类 job.setMapperClass(FlowMapper.class); job.setReducerClass(FlowReducer.class); // 4.设置Map端输出KV类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(FlowBean.class); // 5.设置最终输出KV类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); // 6.设置程序的输入和输出路径 FileInputFormat.setInputPaths(job, new Path("C:\\install\\temp\\input\\input02\\phone_data.txt")); FileOutputFormat.setOutputPath(job, new Path("C:\\install\\temp\\output\\output06")); // 8.指定自定义分区器 job.setPartitionerClass(ProvincePartitioner.class); // 9.同时也指定相应数量的ReduceTask--对应的参数mapreduce.job.reduces，默认为1 job.setNumReduceTasks(5); // 7.提交job boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }

public class WordCountCombiner extends Reducer { private IntWritable outV = new IntWritable(); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable value : values) { sum += value.get(); } outV.set(sum); context.write(key,outV); } }

public class WordCountDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 一获取二关联三设置一提交 // 1.获取配置信息及Job对象 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 2.关联本Driver程序的类 job.setJarByClass(WordCountDriver.class); // 3.关联Mapper和Reducer的业务类 job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); // 4.设置Mapper输出的KV类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5.设置最终输出的KV类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6.设置输入和输出路径 //FileInputFormat.setInputPaths(job, new Path(args[0])); //FileOutputFormat.setOutputPath(job, new Path(args[1])); FileInputFormat.setInputPaths(job, new Path("C:\\install\\temp\\input\\hadoop.txt")); FileOutputFormat.setOutputPath(job, new Path("C:\\install\\temp\\output\\output01-2")); // 8.设置Combiner类--方式一 //job.setCombinerClass(WordCountCombiner.class); // 方式二：其新建的WordCountCombiner的reduce方法处理与正常的WordCountReducer中的reduce方法处理逻辑是一样 // 因此可以直接用此类作为combiner类 job.setCombinerClass(WordCountReducer.class); // 9.设置ReduceTasks的数量--这样就没有reduce阶段，就不会有shuffle，Combiner也就没有用，直接由map输出， // 文件名为part-m-00000，就是不part-r-00000，两者结果是不一样的 // 即如果没有reduce阶段，即使设置了combiner也不起作用 // job.setNumReduceTasks(0); // 7.提交job boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); // 0-正常退出 非0（1）异常终止（结束） } }