HDFS和MapReduce综合实训倒排索引

### HDFS 和 MapReduce 实现倒排索引综合实训 #### 项目概述 本项目旨在通过Hadoop分布式文件系统(HDFS)和MapReduce框架来创建一个倒排索引。此过程涉及设置作业参数、编写映射器(Map)和归约器(Reduce)，以及配置必要的环境变量。 #### 设置Job参数 为了启动MapReduce任务，需指定主类及其依赖组件，包括自定义的`MyMapper`与`MyReducer`类；同时还要指明map阶段产生的键值对类型，reduce阶段处理的数据格式，以及可能存在的combiner逻辑——这里同样采用`MyReducer`实例完成局部聚合操作，并最终确定输入输出路径[^1]。 ```java // Job Configuration Example Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Inverted Index Construction"); job.setJarByClass(InvertedIndexDriver.class); job.setMapperClass(MyMapper.class); // 自定义 Mapper 类 job.setCombinerClass(MyReducer.class); // 使用 Reducer 作为 Combiner job.setReducerClass(MyReducer.class); // 自定义 Reducer 类 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); ``` #### 配置YARN支持 为了让上述程序能够在集群环境中执行，还需调整`etc/hadoop/mapred-site.xml`中的属性，确保MapReduce应用程序能够识别到资源管理平台YARN的存在： ```xml <configuration> <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property> </configuration> ``` 该配置项告知整个计算流程应当依托于YARN来进行调度管理和资源共享[^2]。 #### 编写Mapper函数 在实际编码过程中，`MyMapper`负责解析每条记录并将单词与其所在文档ID关联起来形成初步的词频统计表单。具体来说就是读取文本行，分割成词语列表，再逐一遍历这些词条，发出(word, doc_id)这样的二元组给下游处理器继续加工。 ```java public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().toLowerCase(); // 将整行转换为小写字母 StringTokenizer itr = new StringTokenizer(line); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } ``` #### 编写Reducer函数 另一方面，`MyReducer`则承担着汇总来自不同节点相同关键词计数的任务，即收集所有(key,value)组合后累加得到全局频率分布情况。此外，在某些应用场景下还可以在此环节引入额外过滤条件或者二次排序机制优化性能表现。 ```java public static class MyReducer extends Reducer<Text,IntWritable,Text,IntWritable> { @Override public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException{ int sum = 0; for (IntWritable val : values){ sum += val.get(); } context.write(key,new IntWritable(sum)); } } ```