实验5 MapReduce初级编程实践

1编程实现文件的合并和去重新建java项目添加核心的库创建实现类在本地创建输入文件nano FileA.txtnano FileB.txt创建输入目录将本地文件上传到输入目录中检查是否上传成功确保输出目录不存在编译并运行 MapReduce 程序1: 编译并打包 (导出 JAR)右键点击MapReduceLabs项目。选择Export...选择Java-JAR file。点击Next指定保存的位置JAR file/home/hadoop/lab1.jar点击finish之后如果爆警告可以直接忽略提交 MapReduce Job指令说明查看输出文件目录查看输出结果bug第一次上传文件时只传了文件A没有验证所以疏忽了最后的输出结果和文件A完全一样在第一次输入有误之后调整了代码和输入文件结果输出还是有误而且混乱是没注意删除掉之前的输出目录2编程实现对输入文件的排序Map 阶段 (映射): 我们的Mapper会逐行读取所有输入文件。它读到一行文本例如33。它必须将这个文本 (Text)转换成一个整数 (IntWritable)。它将这个整数作为 Key (键)输出。Mapper输出(Key: 33, Value: NullWritable)。 (Value 是空值不重要)。Shuffle 阶段 (混洗与排序): 这是 Hadoop 的核心。Hadoop 会自动收集所有 Mapper 的输出并对所有的 Key进行排序。Hadoop 看到(33, null),(12, null),(4, null),(1, null)...它会自动将它们排序列为(1, [null]),(4, [null]),(12, [null]),(33, [null])...Reduce 阶段 (规约) - 关键技巧挑战我们如何给一个全局排序的列表添加排名 (1, 2, 3...)? 如果我们有多个 Reducer每个 Reducer 都不知道其他 Reducer 处理了多少数据它们都会从 1 开始排名导致排名错误。解决方案我们强制Hadoop只使用一个 Reducer(job.setNumReduceTasks(1))。结果这一个 Reducer 会按顺序接收到所有排好序的 Key。Reducer 逻辑在 Reducer 内部设置一个计数器private long rank 1;Reducer 收到第一个 Key(Key: 1, Values: [null])。它输出(Rank: 1, Value: 1)。然后rank增加到 2。Reducer 收到第二个 Key(Key: 4, Values: [null])。它输出(Rank: 2, Value: 4)。然后rank增加到 3。以此类推...java实现在之前的包里新建实现类package com.lab1; // 假设我们使用和Dedupe相同的包 import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * MapReduce 程序用于对多个文件中的所有整数进行全局排序,并添加位次(排名)。 */ public class SortAndRank { /** * Mapper (映射器) * * 1. Object, Text, IntWritable, NullWritable * - 输入 Key: Object (行的偏移量, 我们不用) * - 输入 Value: Text (Hadoop 读取的每一行文本, 例如 33) * - 输出 Key: IntWritable (我们把 33 转换成整数 33 作为 Key) * - 输出 Value: NullWritable (空值, 仅作为占位符) */ public static class SortMapper extends MapperObject, Text, IntWritable, NullWritable { private IntWritable numberKey new IntWritable(); // 实例化的 Key Override public void map(Object key, Text value, Context context) throws IOException, InterruptedException { try { // 1. 将输入的 Text (例如 33) 转换成 String, 再转换成 int int number Integer.parseInt(value.toString()); // 2. 将 int 存入我们的 IntWritable Key 中 numberKey.set(number); // 3. 输出 (Key: 33, Value: null) context.write(numberKey, NullWritable.get()); } catch (NumberFormatException e) { // 如果某一行不是数字, 忽略这一行 System.err.println(WARN: Skipping non-integer line: value.toString()); } } } /** * Reducer (规约器) * * 1. IntWritable, NullWritable, LongWritable, IntWritable * - 输入 Key: IntWritable (来自 Shuffle 的排好序的 Key, 例如 1, 4, 5...) * - 输入 Value: IterableNullWritable (一个 null 列表) * - 输出 Key: LongWritable (我们的排名, 1, 2, 3...) * - 输出 Value: IntWritable (排好序的 Key) */ public static class RankReducer extends ReducerIntWritable, NullWritable, LongWritable, IntWritable { // 定义一个 LongWritable 来存储排名, 初始值为 1 private LongWritable rank new LongWritable(1); Override public void reduce(IntWritable key, IterableNullWritable values, Context context) throws IOException, InterruptedException { // key 是排好序的数字 (例如 1) // rank 是当前的排名 (例如 1) // 1. 输出 (Key: 1, Value: 1) context.write(rank, key); // 2. 将排名1, 为下一个数字做准备 rank.set(rank.get() 1); } } /** * Driver (驱动程序) * * 设置并运行 MapReduce Job */ public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, Sort and Rank); job.setJarByClass(SortAndRank.class); job.setMapperClass(SortMapper.class); job.setReducerClass(RankReducer.class); // --- 关键设置 --- // 1. (重要!) 设置 Reduce 任务的数量为 1, 以确保全局排序和排名 job.setNumReduceTasks(1); // 2. 设置 Mapper 的输出 Key/Value 类型 (必须匹配 Mapper) job.setMapOutputKeyClass(IntWritable.class); job.setMapOutputValueClass(NullWritable.class); // 3. 设置 Reducer (即最终) 的输出 Key/Value 类型 (必须匹配 Reducer) job.setOutputKeyClass(LongWritable.class); job.setOutputValueClass(IntWritable.class); // 4. 检查参数 if (args.length 2) { System.err.println(用法: SortAndRank HDFS输入目录 HDFS输出目录); System.exit(2); } // 5. 设置 HDFS 输入路径 (args[0]) FileInputFormat.addInputPath(job, new Path(args[0])); // 6. 设置 HDFS 输出路径 (args[1]) FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7. 提交 Job 并等待完成 boolean success job.waitForCompletion(true); System.exit(success ? 0 : 1); } }创建输入文件nano File1.txtnano File2.txtnano File3.txt在hdfs中创建输入目录上传文件到hdfs编译并导出jar包运行jar包查看输出结果3对给定的表格进行信息的排序挑战我们的目标是找到一个三代关系A - B - C(孙辈 - 父母 - 祖辈)。A是grandchild(孙辈)B是parent(父母)C是grandparent(祖辈)我们的输入文件只包含A - B和B - C这两种关系并且它们都混在一个文件里格式都是child parent。Join (连接) 键要把(A, B)和(B, C)这两个关系连接起来共同的“桥梁”是B(父母)。因此我们的Join 键必须是B。Map 阶段 (映射 - 关键技巧):Mapper 将逐行读取输入文件child parent。对于每一行它实际上代表了两种不同的信息我们必须用两种方式把它发送出去方式 1将其视为 A - B (孙辈 - 父母)当 Mapper 读到Steven Lucy它会想“Steven是一个孩子他的父母是Lucy。我需要用Lucy作为 Key 去找Lucy的父母。”它发送(Key: Lucy, Value: 1-Steven)(这里的 1- 是一个标签意思是 这是一个孩子/孙辈)方式 2将其视为 B - C (父母 - 祖辈)当 Mapper 读到Steven Lucy它会想“Steven是一个父母他的孩子是...等等这个逻辑不对”让我们换一种更简单的逻辑当 Mapper 读到Steven Lucy它代表Steven是孩子Lucy是父母。当 Mapper 读到Lucy Mary它代表Lucy是孩子Mary是父母。Join 键应该是Lucy。Mapper 逻辑 (更正后)读取Steven Lucy我们需要用Lucy作为 Key。我们需要标记Steven是Lucy的孩子。Mapper 发送(Key: Lucy, Value: CHILD-Steven)读取Lucy Mary我们需要用Lucy作为 Key。我们需要标记Mary是Lucy的父母。Mapper 发送(Key: Lucy, Value: PARENT-Mary)Shuffle 阶段 (混洗与排序):Hadoop 自动工作把所有相同的 KeyLucy分组到一起。Reduce 阶段 (规约):Reducer 会收到一个键和一组值。例如(Key: Lucy, Values: [CHILD-Steven, CHILD-Jone, PARENT-Mary, PARENT-Frank])Reducer 逻辑:创建两个空列表grandchildren和grandparents。遍历Values列表遇到CHILD-Steven- 把Steven添加到grandchildren列表。遇到CHILD-Jone- 把Jone添加到grandchildren列表 。遇到PARENT-Mary- 把Mary添加到grandparents列表 。遇到PARENT-Frank- 把Frank添加到grandparents列表 。当两个列表都填充完毕后执行一个双重循环 (笛卡尔积)for (grandchild : grandchildren):for (grandparent : grandparents):输出(Key: grandchild, Value: grandparent)结果(Steven, Mary)(Steven, Frank)(Jone, Mary)(Jone, Frank)java代码实现新建实现类package com.lab1; // 假设我们使用和Dedupe相同的包 import java.io.IOException; import java.util.ArrayList; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * MapReduce 程序用于挖掘 child-parent 关系 * 并输出 grandchild-grandparent 关系。 */ public class Grandparent { // 定义标签用于在 Reducer 中区分 private static final String CHILD_TAG CHILD-; private static final String PARENT_TAG PARENT-; /** * Mapper (映射器) * * 1. Object, Text, Text, Text * - 输入 Key: Object (行的偏移量, 我们不用) * - 输入 Value: Text (Hadoop 读取的每一行, 例如 Steven Lucy) * - 输出 Key: Text (Join 键, 即 parent 或 child) * - 输出 Value: Text (带标签的值, 例如 CHILD-Steven) */ public static class JoinMapper extends MapperObject, Text, Text, Text { private Text joinKey new Text(); private Text outputValue new Text(); Override public void map(Object key, Text value, Context context) throws IOException, InterruptedException { // 将一行 child parent 拆开 String[] parts value.toString().split(\\s); // 按一个或多个空格拆分 // 确保数据格式正确 (必须是两列) if (parts.length 2) { String child parts[0]; String parent parts[1]; // 忽略标题行 child parent if (child.equals(child) parent.equals(parent)) { return; } // --- 关键逻辑 --- // 1. 发送 孩子 记录 // Join 键是 parent (Lucy) // Value 是带标签的 child (CHILD-Steven) joinKey.set(parent); outputValue.set(CHILD_TAG child); context.write(joinKey, outputValue); // 2. 发送 父母 记录 // Join 键是 child (Steven) // Value 是带标签的 parent (PARENT-Lucy) joinKey.set(child); outputValue.set(PARENT_TAG parent); context.write(joinKey, outputValue); } } } /** * Reducer (规约器) * * 1. Text, Text, Text, Text * - 输入 Key: Text (Join 键, 例如 Lucy) * - 输入 Value: IterableText (标签列表, [CHILD-Steven, PARENT-Mary, ...]) * - 输出 Key: Text (Grandchild, 孙辈) * - 输出 Value: Text (Grandparent, 祖辈) */ public static class JoinReducer extends ReducerText, Text, Text, Text { private Text grandchild new Text(); private Text grandparent new Text(); Override public void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { // Key 是 B (例如 Lucy) ListString children new ArrayList(); // 孙辈 (A) ListString parents new ArrayList(); // 祖辈 (C) // 1. 遍历所有值根据标签分类 for (Text val : values) { String v val.toString(); if (v.startsWith(CHILD_TAG)) { children.add(v.substring(CHILD_TAG.length())); // 添加 Steven } else if (v.startsWith(PARENT_TAG)) { parents.add(v.substring(PARENT_TAG.length())); // 添加 Mary } } // 2. 如果 Join 键 (Lucy) 既有孩子 (Steven)又有父母 (Mary) if (!children.isEmpty() !parents.isEmpty()) { // 3. 执行笛卡尔积输出所有组合 for (String c : children) { grandchild.set(c); // e.g., Steven for (String p : parents) { grandparent.set(p); // e.g., Mary context.write(grandchild, grandparent); // 输出 (Steven, Mary) } } } } } /** * Driver (驱动程序) * * 设置并运行 MapReduce Job */ public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, Grandparent Join); job.setJarByClass(Grandparent.class); job.setMapperClass(JoinMapper.class); job.setReducerClass(JoinReducer.class); // --- 关键设置 --- // 1. Mapper 的输出和 Reducer 的输出类型不同 // 所以必须明确设置 Mapper 的输出类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); // 2. Reducer (即最终) 的输出 Key/Value 类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 3. 检查参数 if (args.length 2) { System.err.println(用法: Grandparent HDFS输入目录 HDFS输出目录); System.exit(2); } // 4. 设置 HDFS 输入路径 (args[0]) FileInputFormat.addInputPath(job, new Path(args[0])); // 5. 设置 HDFS 输出路径 (args[1]) FileOutputFormat.setOutputPath(job, new Path(args[1])); // 6. 提交 Job 并等待完成 boolean success job.waitForCompletion(true); System.exit(success ? 0 : 1); } }创建本地输入文件建立输入目录将文件复制到输入目录中编译并打包 (导出 JAR)右键整个项目运行jar包查看结果