java操作Hadoop的核心类库MapReduce

傷城~ 2021-10-23 09:21 371阅读 0赞

最近国网这边要求使用大数据框架写项目，无奈之前对于Hadoop这些一无所知，只能先学习一下开发时主要用到的一些技术点，后续有空再继续补充原理知识

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]  
MapReduce主要是采用分而治之的思想，上图的理解主要如下。Hadoop的Hdfs分布式存储系统将大数据分成很多数据块，每个数据块（input splits）传入到input阶段，

**MapReduce数据块处理**  
输入拆分：  
输入到MapReduce工作被划分成固定大小的块叫做 input splits ，输入折分是由单个映射消费输入块。  
映射 - Mapping  
这是在 map-reduce 程序执行的第一个阶段。在这个阶段中的每个分割的数据被传递给映射函数来产生输出值。  
重排(shuffling)  
这个阶段消耗映射阶段的输出。它的任务是合并映射阶段输出的相关记录。在我们的例子，同样的词汇以及它们各自出现频率。（可以理解为将相同key的map整理成一个新的map（value是个数组））  
Reducing  
在这一阶段，从重排阶段输出值汇总。这个阶段结合来自重排阶段值，并返回一个输出值。总之，这一阶段汇总了完整的数据集。

# java操作的核心方法 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]  
**Maping阶段**  
首先导入Hadoop的核心类Mapper类，里面有四个方法，先讨论常用的map方法

public class TestReduceMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line =key.toString();//这个值是Hadoop自己产生的编号
            String year =value.toString();
            context.write(new Text(year),new IntWritable());//类型要和泛型声明一致
        }
    }

主要是实现 org.apache.hadoop.mapreduce.Mapper类，重写map方法  
将传入的数据块的key-value对处理完之后，使用context.write方法重新返回一个key-value对，然后之后会由重排阶段将每个数据块处理后的key-value对进行整合

**Reduce阶段**

public class ReduceMapper extends Reducer<Text, IntWritable, Text, IntWritable> {
        /**
         * 从重排阶段输出值汇总。这个阶段结合来自重排阶段值，并返回一个输出值。
         * 总之，这一阶段汇总了完整的数据集。
         * 这个阶段的数据类型要和重排得到的格式相同，即数组类型
         *
         */
        @Override
        protected void reduce(Text keyIn, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int maxValue = Integer.MIN_VALUE;
            for (IntWritable value:values ) {
              maxValue = Math.max(maxValue,value.get());//比较获取最大值
            }
           context.write(new Text(keyIn),new IntWritable(maxValue));
        }
    }

**发布作业阶段**  
此处写一个main函数，主要用来运行作业

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    		Configuration conf = new Configuration();
    		
    		// 获得一个job对象，用来完成一个mapreduce作业
    		Job job = Job.getInstance(conf);	
    		//设置作业名称，便于调试
    		job.setJarByClass("testTask");
    		// 让程序找到主入口，即main函数所在类
    		job.setJarByClass(WordCountDriver.class);
    		// 添加目标数据文件输入路径，可以添加多个
    	   //也可以是文件夹，但子目录不会递归
    		FileInputFormat.addInputPath(job, new Path(args[0]));
    		//设置输出路径，只能一个，而且不能存在（会动态创建）
    		FileOutputFormat.setOutputPath(job, new Path(args[1]));
    		
    		// 设置调用的map方法和reduce方法
    		job.setMapperClass(WordCountMap.class);
    		job.setReducerClass(WordCountReducer.class);
    		
    		// 指定map输出键值对的类型
    		job.setMapOutputKeyClass(Text.class);
    		job.setMapOutputValueClass(IntWritable.class);
    		
    		// 指定reduce输出键值对的类型
    		job.setOutputKeyClass(Text.class);
    		job.setOutputValueClass(IntWritable.class);
    		
    		// 提交job任务
    		boolean result = job.waitForCompletion(true);
    		
    		
    	}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70]: /images/20211023/00a50c7b11a94d42878a6a38913c3d63.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzAwMDMw_size_16_color_FFFFFF_t_70 1]: /images/20211023/444b4b86e6f74f9c912ef9464aca0123.png