如何在map()与reduce()间共享HashMap及reduce()间共享整数变量？

看起来你需要在map和reduce环节之间维护全局状态，同时还要跟踪计数器并最终输出最大值。根据你使用的是单机Java Stream还是分布式Hadoop MapReduce，我给你两种针对性的解决方案：

一、单机场景（Java Stream API） #

如果你是用Java的Stream来实现map-reduce逻辑，线程安全是关键——毕竟Stream可能并行执行。我们可以封装一个共享状态类来管理HashMap和计数器：

1. 定义线程安全的共享状态类 #

这个类把HashMap、计数器和最大值逻辑封装在一起，保证多线程下的安全性：

public class SharedState {
    // 线程安全的HashMap存储key-value
    private final ConcurrentHashMap<String, Integer> keyValueMap = new ConcurrentHashMap<>();
    // 原子计数器，保证线程安全的增减
    private final AtomicInteger processedCount = new AtomicInteger(0);
    // 跟踪当前最大值及对应key（volatile保证多线程可见性）
    private volatile int currentMaxValue = Integer.MIN_VALUE;
    private volatile String currentMaxKey;

    // 更新map并同步计数器、最大值
    public void update(String key, int value) {
        keyValueMap.put(key, value);
        processedCount.incrementAndGet();
        
        // 检查是否更新最大值
        if (value > currentMaxValue) {
            currentMaxValue = value;
            currentMaxKey = key;
        }
    }

    // 判断是否处理完所有key-value
    public boolean isLastEntry() {
        return processedCount.get() == keyValueMap.size();
    }

    // 获取最大值的key-value字符串
    public String getMaxEntry() {
        return String.format("%s:%d", currentMaxKey, currentMaxValue);
    }
}

2. 在Stream流水线中使用共享状态 #

把SharedState作为全局状态，在map阶段更新，reduce阶段判断并输出：

public static void main(String[] args) {
    // 初始化共享状态
    SharedState state = new SharedState();
    
    // 模拟输入数据
    List<String> inputData = Arrays.asList("a:12", "b:45", "c:30", "d:60");

    input.stream()
        // map阶段：解析数据并更新共享状态
        .map(entryStr -> {
            String[] parts = entryStr.split(":");
            String key = parts[0];
            int value = Integer.parseInt(parts[1]);
            state.update(key, value);
            return new AbstractMap.SimpleEntry<>(key, value);
        })
        // reduce阶段：对比当前entry与全局最大值，最后一组时输出
        .reduce((prevEntry, currEntry) -> {
            if (state.isLastEntry()) {
                System.out.println("全局最大值：" + state.getMaxEntry());
            }
            return currEntry;
        });
}

并行Stream下也能保证安全，因为我们用了ConcurrentHashMap和AtomicInteger来规避并发问题。

二、分布式场景（Hadoop MapReduce） #

如果是分布式的Hadoop MapReduce，普通内存共享行不通，得用Hadoop提供的分布式机制：

1. 共享HashMap：用DistributedCache #

适合小规模的HashMap，先把HashMap序列化到文件上传到HDFS，再让所有Task通过DistributedCache读取：

// Job提交前：序列化HashMap并上传到HDFS
HashMap<String, Integer> initMap = new HashMap<>();
ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream("shared_map.ser"));
oos.writeObject(initMap);
oos.close();

FileSystem fs = FileSystem.get(new Configuration());
fs.copyFromLocalFile(new Path("shared_map.ser"), new Path("/tmp/shared_map.ser"));

// 配置DistributedCache
DistributedCache.addCacheFile(new Path("/tmp/shared_map.ser").toUri(), job.getConfiguration());

在Mapper/Reducer的setup()方法中读取：

private HashMap<String, Integer> sharedMap;

@Override
protected void setup(Context context) throws IOException, InterruptedException {
    Path[] cacheFiles = DistributedCache.getLocalCacheFiles(context.getConfiguration());
    ObjectInputStream ois = new ObjectInputStream(new FileInputStream(cacheFiles[0].toString()));
    sharedMap = (HashMap<String, Integer>) ois.readObject();
    ois.close();
}

如果需要实时更新全局HashMap，建议结合HDFS加锁（比如用FileSystem.create()的独占模式），但会影响性能——更推荐用MapReduce的原生聚合逻辑替代全局HashMap的维护。

2. 共享整数计数器：用Hadoop Counter #

Hadoop的Counter是全局可见的，所有Task都能更新和读取：

// 在Reducer中更新计数器
context.getCounter("CustomCounters", "TOTAL_PROCESSED").increment(1);

// 获取当前计数器值
long total = context.getCounter("CustomCounters", "TOTAL_PROCESSED").getValue();

判断最后一组时，可以在Reducer的cleanup()方法中检查计数器值是否等于预期的总key数，然后输出最大值。

3. 更高效的全局最大值实现 #

其实在Hadoop中，找全局最大值不用维护全局HashMap，更符合设计的方式是：

1. Mapper输出每个key-value对
2. 第一个Reducer计算每个key的最大值，输出<"max", value>
3. 用一个单节点Reducer汇总所有"max"值，得到全局最大值

这种方式避免了分布式共享状态的并发问题，性能更优。

内容的提问来源于stack exchange，提问作者Varun Gawande