java hadoop（四） hadoop mapreduce 分区案例

水深无声 2022-10-05 13:57 217阅读 0赞

我们在编码步骤那章简单介绍了一下分区。

其实通俗来说，分区就是根绝K的某个特征属性（可能是大小、长度以及尾号等等），将K的数据写到不同的文件里。

比如我们根据ID来分区，我们可以根据尾号来分10个文件。

有几个分区就有几个Reduce Task。

**实现分区案例：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

我们根据hdfs文件中的这个字段大小进行分区，大于15的一个区，小于15的一个区，最终有两个out文件。

### **目录结构：** ###

**![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]**

### **Mapper：** ###

package partitiondemo;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    
    /**
     * 传入四个参数，第一个参数是这一行的起始点在文件中的偏移量 第二个就是读取到的内容
     * 第三个 第四个 是我们输出的K2 V2类型
     * 思路：将文本内容作为K2。 V2直接为空
     */
    public class MyMapper extends Mapper<LongWritable, Text,Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //value 直接写入空值
            context.write(value, NullWritable.get());
        }
    }

### **Partitioner：** ###

package partitiondemo;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    /**
     * 这里的两个参数就是map的K2 V2
     */
    public class MyPartitioner  extends Partitioner<Text, NullWritable> {
    
        @Override
        public int getPartition(Text text, NullWritable nullWritable, int i) {
            //取出根据数值分区的那个值
            String num = text.toString().split("\t")[5];
    
            //分区
            if(Integer.parseInt(num)>15){
                return 1;
            }else {
                return 0;
            }
        }
    }

### **Reducer：** ###

package partitiondemo;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * 没有处理
     */
    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());
        }
    }

### **Main：** ###

package partitiondemo;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    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;
    
    public class ReducerMain {
    
        public static void main(String[] args) throws Exception{
            //创建job
            Configuration configuration = new Configuration();
            Job job = Job.getInstance(configuration, "my-partition");
            //指定job所在jar包
            job.setJarByClass(ReducerMain.class);
            //指定文件读取方式 路径
            job.setInputFormatClass(TextInputFormat.class);
            TextInputFormat.addInputPath(job, new Path("hdfs://192.168.40.150:9000/test.txt"));
            //指定mapper
            job.setMapperClass(MyMapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
    
            //指定自定义分区类
            job.setPartitionerClass(MyPartitioner.class);
    
            //指定reducer
            job.setReducerClass(MyReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
    
            //设置分区个数 我们是两个reducer
            job.setNumReduceTasks(2);
            //指定输出方式类以及输出路径 目录必须不存在
            job.setOutputFormatClass(TextOutputFormat.class);
            TextOutputFormat.setOutputPath(job, new Path("hdfs://192.168.40.150:9000/lgy_test/res"));
            //将job提交到yarn集群
            boolean bl = job.waitForCompletion(true);
            System.exit(bl?0:1);
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221005/763eb418b7c24c69bbc9d8e7baf7e1d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221005/22a62a263c024900a58ff753fee6014c.png