实验三 MapReduce 序列化数据处理

### MapReduce 实验三：序列化数据处理 #### 背景说明 在分布式计算框架 Hadoop 中，MapReduce 是一种核心的数据处理模型。为了支持高效的大规模数据传输和存储，Hadoop 使用了一种轻量级的二进制序列化机制来替代传统的 Java 序列化[^3]。这种机制不仅提高了性能，还减少了磁盘 I/O 和网络带宽消耗。 以下是基于实验三的具体实现方法以及示例代码： --- #### 1. 自定义 Bean 类实现 Writable 接口 要使自定义对象能够在 MapReduce 过程中被序列化和反序列化，需要让其继承 `Writable` 接口并重写相关方法。以下是一个简单的例子： ```java import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class MovieRating implements Writable { private int userID; // 用户 ID private String itemID; // 商品/电影 ID private double score; // 打分 public MovieRating() {} // 默认构造函数必须存在 public MovieRating(int userID, String itemID, double score) { this.userID = userID; this.itemID = itemID; this.score = score; } @Override public void write(DataOutput out) throws IOException { out.writeInt(userID); out.writeUTF(itemID); out.writeDouble(score); } @Override public void readFields(DataInput in) throws IOException { this.userID = in.readInt(); this.itemID = in.readUTF(); this.score = in.readDouble(); } // Getter and Setter methods (Optional but recommended) public int getUserID() { return userID; } public void setUserID(int userID) { this.userID = userID; } public String getItemID() { return itemID; } public void setItemID(String itemID) { this.itemID = itemID; } public double getScore() { return score; } public void setScore(double score) { this.score = score; } } ``` 此部分实现了自定义对象的序列化逻辑，确保它可以安全地在网络上传输或保存到文件系统中。 --- #### 2. Mapper 的设计与实现 Mapper 的主要职责是从输入数据集中提取有用的信息，并将其转换成键值对的形式输出。假设我们有一个评分数据集，每条记录包含用户 ID (`userID`)、商品/电影 ID (`itemID`) 及打分 (`score`)。 ```java import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import java.io.IOException; public class RatingMapper extends Mapper<LongWritable, Text, IntWritable, MovieRating> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split(","); if (fields.length >= 3) { try { int userID = Integer.parseInt(fields[0]); String itemID = fields[1]; double score = Double.parseDouble(fields[2]); MovieRating rating = new MovieRating(userID, itemID, score); context.write(new IntWritable(userID), rating); // 输出 Key: UserID Value: MovieRating Object } catch (NumberFormatException e) { System.err.println("Invalid input data format"); } } } } ``` 在此过程中，我们将原始数据解析为 `(userID, MovieRating)` 键值对形式。 --- #### 3. Reducer 的设计与实现 Reducer 的作用是对相同键（这里是 `userID`）的所有值进行聚合操作。例如，在本案例中，我们需要统计每个用户的总评分及其对应的电影列表。 ```java import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import java.io.IOException; import java.util.ArrayList; import java.util.List; public class RatingReducer extends Reducer<IntWritable, MovieRating, Text, Text> { @Override protected void reduce(IntWritable key, Iterable<MovieRating> values, Context context) throws IOException, InterruptedException { List<String> itemsWithScores = new ArrayList<>(); double totalScore = 0.0; for (MovieRating rating : values) { StringBuilder sb = new StringBuilder(rating.getItemID()); sb.append(",").append(rating.getScore()); // 构造 ItemID,score 形式的字符串 itemsWithScores.add(sb.toString()); totalScore += rating.getScore(); // 计算总分数 } StringBuilder resultBuilder = new StringBuilder(); for (String itemWithScore : itemsWithScores) { resultBuilder.append(itemWithScore).append(";"); // 合并多个 ItemID+score } String outputKey = "User-" + key.toString(); String outputValue = resultBuilder.toString(); context.write(new Text(outputKey), new Text(outputValue)); // 输出最终结果 } } ``` 这里将同一用户的评分汇总，并按指定格式输出。 --- #### 4. Job 配置 最后一步是配置整个 MapReduce 作业流程。这包括设置输入路径、输出路径、Mapper 和 Reducer 类型等参数。 ```java import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class MovieRecommendationJob { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Movie Recommendation"); job.setJarByClass(MovieRecommendationJob.class); job.setMapperClass(RatingMapper.class); job.setCombinerClass(RatingReducer.class); // Optional combiner to optimize performance job.setReducerClass(RatingReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } } ``` 以上代码片段展示了如何构建完整的 MapReduce 流水线。 --- ### 总结 通过上述步骤，可以完成一个基本的 MapReduce 数据处理任务。其中涉及到了自定义对象的序列化、Mapper 和 Reducer 的具体实现以及整体作业的配置过程[^1][^2][^3]。 ---