引言

在当前数据驱动的应用场景中，高效处理TB乃至PB级数据已成为系统设计的关键环节。Hadoop 和 Apache Spark 作为主流的大数据处理框架，凭借其分布式架构和可扩展性，被广泛应用于日志分析、推荐系统、实时计算等领域。本文将聚焦于如何通过 Java 编程语言，结合 Hadoop 的 MapReduce 模型与 Spark 的内存计算引擎，完成典型的大规模数据处理任务。

Hadoop 分布式处理详解

Hadoop 是一个支持高吞吐量数据处理的开源框架，核心由 HDFS（Hadoop Distributed File System）和 MapReduce 计算模型构成。HDFS 负责将大文件切片并分布存储在集群节点上，而 MapReduce 则提供了一种编程范式，用于并行处理这些分片数据。

要在 Java 环境中开发 Hadoop 应用，需引入 Hadoop 客户端依赖，并配置集群连接参数。以下是一个基于 MapReduce 实现词频统计的完整示例：

package com.example.bigdata;

import org.apache.hadoop.conf.Configured;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordFrequencyJob extends Configured implements Tool {

    public static class WordMapper extends Mapper<Object, Text, Text, IntWritable> {
        private Text outputKey = new Text();
        private IntWritable oneValue = new IntWritable(1);

        @Override
        protected void map(Object key, Text value, Context context) 
                throws IOException, InterruptedException {
            StringTokenizer tokenizer = new StringTokenizer(value.toString());
            while (tokenizer.hasMoreTokens()) {
                String token = tokenizer.nextToken().toLowerCase().replaceAll("[^a-z]", "");
                if (!token.isEmpty()) {
                    outputKey.set(token);
                    context.write(outputKey, oneValue);
                }
            }
        }
    }

    public static class SumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable totalCount = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) 
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            totalCount.set(sum);
            context.write(key, totalCount);
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf(), "word frequency counter");
        job.setJarByClass(WordFrequencyJob.class);

        job.setMapperClass(WordMapper.class);
        job.setCombinerClass(SumReducer.class);
        job.setReducerClass(SumReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job, new Path(args[0]));
        TextOutputFormat.setOutputPath(job, new Path(args[1]));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new WordFrequencyJob(), args);
        System.exit(exitCode);
    }
}

该程序继承 Configured 并实现 Tool 接口，便于命令行参数解析与配置管理。Map 阶段对每行文本进行分词并输出 (word, 1) 对；Reduce 阶段汇总相同单词的计数。本地运行时可通过打包为 JAR 文件提交至 Hadoop 集群执行。

Spark 内存计算实践

Apache Spark 提供了比 Hadoop 更高效的计算能力，尤其适合迭代算法和交互式查询。其核心抽象是弹性分布式数据集（RDD），支持在内存中缓存中间结果，避免频繁磁盘 I/O。

使用 Spark 的 Java API 可以简洁地表达数据转换逻辑。以下是等效的词频统计实现：

package com.example.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.regex.Pattern;

public class SparkWordCounter {

    private static final Pattern SPACE_PATTERN = Pattern.compile("\\s+");

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
            .setAppName("Spark Word Count Example")
            .setMaster("local[4]"); // 使用本地四线程模式测试

        try (JavaSparkContext sc = new JavaSparkContext(conf)) {
            JavaRDD<String> lines = sc.textFile(args[0]);

            JavaPairRDD<String, Integer> wordCounts = lines
                .flatMap(line -> Arrays.asList(SPACE_PATTERN.split(line.toLowerCase())).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1))
                .reduceByKey(Integer::sum);

            wordCounts.saveAsTextFile(args[1]);
        }
    }
}

上述代码利用 flatMap 将每一行拆分为单词流，再通过 mapToPair 构造键值对，最后调用 reduceByKey 合并相同键的值。整个流程链式调用，语义清晰且易于维护。生产环境中可将 master 设置为 spark://host:port 连接独立集群。

框架对比与选型建议

• Hadoop MapReduce：适用于对延迟不敏感的离线批处理任务，如月度报表生成、历史数据归档等。其强一致性与容错机制保障了大规模作业的稳定性。
• Apache Spark：更适合需要快速响应的场景，包括机器学习训练、流式处理（Structured Streaming）、图计算等。由于支持 DAG 执行引擎和内存缓存，性能通常优于传统 MapReduce 数倍。

两者均可通过 Java API 无缝集成到企业现有系统中。实际项目中也可采用"Hadoop 存储 + Spark 计算"的混合架构，即使用 HDFS 作为统一数据湖，Spark 作为上层计算引擎，实现存储与计算分离的最佳实践。