2.2.9 hadoop体系之离线计算-mapreduce分布式计算-流量统计之手机号码分区

素颜马尾好姑娘i 2022-12-13 09:55 129阅读 0赞

**目录**

1.需求分析

2.代码方案

2.1 自定义分区

2.2 JobMain添加分区设置

2.3 分区结果

--------------------

# 1.需求分析 #

在需求一的基础上，继续完善，将不同的手机号分到不同的数据文件的当中去，需要自定 义分区来实现，这里我们自定义来模拟分区，将以下数字开头的手机号进行分开

*  135 开头的数据到一个分区文件
 *  136开头的数据到一个分区文件
 *  137开头的数据到一个分区文件
 *  其他数据到一个分区文件

# 2.代码方案 #

## 2.1 自定义分区 ##

package ucas.mapreduce_flowcount_partitioner;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class FlowPartition extends Partitioner<Text, FlowBean> {
        @Override
        public int getPartition(Text text, FlowBean flowBean, int i) {
            String line = text.toString();
            if (line.startsWith("135")) {
                return 0;
            } else if (line.startsWith("136")) {
                return 1;
            } else if (line.startsWith("137")) {
                return 2;
            } else {
                return 3;
            }
        }
    }

## 2.2 JobMain添加分区设置 ##

job.setPartitionerClass(FlowPartition.class);

package ucas.mapreduce_flowcount_partitioner;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    public class JobMain extends Configured implements Tool {
        @Override
        public int run(String[] strings) throws Exception {
            //创建一个任务对象
            Job job = Job.getInstance(super.getConf(), "mapreduce_flowcount");
    
            //打包放在集群运行时，需要做一个配置
            job.setJarByClass(JobMain.class);
            //第一步:设置读取文件的类: K1 和V1
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job, new Path("hdfs://192.168.0.101:8020/input/flowcount"));
    
            //第二步：设置Mapper类
            job.setMapperClass(FlowCountMapper.class);
            //设置Map阶段的输出类型: k2 和V2的类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
    
            //第三,四，五，六步采用默认方式(分区，排序，规约，分组)
            job.setPartitionerClass(FlowPartition.class);
    
            //第七步 ：设置文的Reducer类
            job.setReducerClass(FlowCountReducer.class);
            //设置Reduce阶段的输出类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            //设置Reduce的个数
    
            //第八步:设置输出类
            job.setOutputFormatClass(TextOutputFormat.class);
            //设置输出的路径
            TextOutputFormat.setOutputPath(job, new Path("hdfs://192.168.0.101:8020/out/flowcount_out"));
    
    
            boolean b = job.waitForCompletion(true);
            return b ? 0 : 1;
    
        }
    
        public static void main(String[] args) throws Exception {
            Configuration configuration = new Configuration();
    
            //启动一个任务
            int run = ToolRunner.run(configuration, new JobMain(), args);
            System.exit(run);
        }
    
    }

## 2.3 分区结果 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221123/9b5788c2cd664677a08f69ee9f76a50f.png