如何使用Hadoop MapReduce Replicated Join实现两数据集内连接

1. 复制数据集的存储与分发 #

直接将小于1GB的小数据集上传到分布式存储（如HDFS）的公共可读路径，提交任务时通过-files参数指定该文件路径，框架会自动把文件分发到所有Map任务的本地工作目录，无需走Shuffle流程。

2. Setup阶段读取逻辑与数据结构选择 #

在Map任务的setup方法中读取本地的小数据集，不要使用普通单值哈希表，改用多值哈希表存储，解决重复键覆盖问题：

• 键：复制数据集的第1列（关联键）
• 值：该关联键对应的所有复制数据集第2列的列表
  示例伪代码如下：

// 初始化多值哈希表
HashMap<String, List<String>> smallTable = new HashMap<>();
// 逐行读取复制数据集
BufferedReader reader = new BufferedReader(new FileReader("./small_dataset.txt"));
String line;
while ((line = reader.readLine()) != null) {
    String[] cols = line.split(" ");
    String joinKey = cols[0];
    String smallVal = cols[1];
    // 键不存在时先初始化列表
    if (!smallTable.containsKey(joinKey)) {
        smallTable.put(joinKey, new ArrayList<>());
    }
    // 同键值追加到列表，不会覆盖已有数据
    smallTable.get(joinKey).add(smallVal);
}
reader.close();

3. Map阶段关联实现逻辑 #

因为已经设置Reducer数量为0，所有关联逻辑直接在Map方法中完成，主数据集作为Map的输入源，逐行处理逻辑如下：

1. 拆分当前主数据集行，取第2列作为关联键
2. 到多值哈希表中查询是否存在该关联键，不存在直接跳过
3. 存在的话遍历该键对应的所有值列表，逐行拼接输出
   示例伪代码如下：

// 处理主数据集的每一行
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String[] cols = value.toString().split(" ");
    String mainFirstCol = cols[0];
    String joinKey = cols[1]; // 主数据集第2列为关联键
    if (smallTable.containsKey(joinKey)) {
        // 遍历同键所有小表值，每条都输出避免数据丢失
        for (String smallSecondCol : smallTable.get(joinKey)) {
            // 拼接规则可按需求调整，当前适配你给出的预期输出格式
            context.write(null, new Text(joinKey + " " + smallSecondCol + " " + mainFirstCol));
        }
    }
}

注意事项 #

• 内存配置：小数据集全量加载到Map内存，需要确保Map任务的内存配额≥小数据集的2倍，1GB的小数据集建议设置Map内存为2GB以上
• 数据预处理：如果小数据集有冗余行，可在读取阶段提前做去重，减少内存占用
• 提交参数：Hadoop任务提交时需添加-D mapreduce.job.reduces=0参数关闭Reducer，添加-files hdfs://你的小数据集路径完成文件分发

内容的提问来源于stack exchange，提问作者Marc-9