当前位置：首页 > news >正文

Spark 和 Hadoop MapReduce 的基本概念及区别

news 2025/7/15 5:56:58

1. Apache Spark

Spark 是一个开源的集群计算框架，专为快速数据处理而设计。它最初由 UC Berkeley 的 AMPLab 开发，后来成为 Apache 软件基金会的顶级项目。Spark 的核心特点包括：

内存计算：数据可以缓存在内存中，大幅提升迭代计算和交互式查询的速度。
多语言支持：支持 Java、Scala、Python（PySpark）、R 和 SQL。
丰富的组件栈：包含 Spark SQL（结构化数据处理）、MLlib（机器学习）、GraphX（图计算）和 Spark Streaming（实时流处理）。
DAG 调度引擎：通过有向无环图（DAG）优化任务执行，减少数据冗余传输。

2. Hadoop MapReduce

Hadoop MapReduce 是 Apache Hadoop 的核心组件之一，是一种批处理计算模型。它将计算任务分解为两个主要阶段：

Map 阶段：将输入数据分割并转换为键值对。
Reduce 阶段：对 Map 输出的键值对进行汇总和聚合。
Hadoop MapReduce 的设计理念是 "移动计算而非数据"，适合处理大规模数据集，但它的主要局限在于每次计算都需要读写磁盘，导致迭代计算效率较低。

Spark 与 Hadoop MapReduce 的核心区别

特性	Apache Spark	Hadoop MapReduce
计算模式	内存计算为主，支持迭代和流式处理	磁盘 I/O 密集，批处理为主
处理速度	比 MapReduce 快 10-100 倍（内存中）	较慢，每次任务都需磁盘读写
编程模型	RDD（弹性分布式数据集）、DataFrame、Dataset	严格的 Map 和 Reduce 函数
适用场景	迭代计算（如机器学习）、实时流处理、交互式分析	批处理、ETL、大规模数据处理
数据存储	可连接 HDFS、S3、HBase 等，但自身不提供存储	依赖 HDFS（Hadoop 分布式文件系统）
任务调度	DAG 调度器优化执行路径，减少 Shuffle 开销	简单的 Map 和 Reduce 阶段，Shuffle 开销大
延迟	低延迟（秒级响应），适合交互式查询	高延迟（分钟级），启动开销大
语言支持	多语言（Java、Scala、Python、R、SQL）	主要支持 Java，其他语言需额外包装
组件生态	集成 Spark SQL、MLlib、GraphX、Spark Streaming	依赖 Hive（SQL）、Pig（数据流）、Mahout（机器学习）

代码对比示例

Spark（Python 版）

下面是一个使用 PySpark 计算单词频率的示例：

from pyspark.sql import SparkSession# 创建SparkSession
spark = SparkSession.builder.appName("WordCount").getOrCreate()# 读取文本文件
lines = spark.read.text("input.txt").rdd.map(lambda r: r[0])# 分词并计数
counts = lines.flatMap(lambda x: x.split(" ")) \.map(lambda x: (x, 1)) \.reduceByKey(lambda a, b: a + b)# 输出结果
counts.saveAsTextFile("output")# 停止SparkSession
spark.stop()

Hadoop MapReduce（Java 版）

对应的 Hadoop MapReduce 实现：

import java.io.IOException;
import java.util.StringTokenizer;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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class WordCount {public static class TokenizerMapperextends Mapper<Object, Text, Text, IntWritable>{private final static IntWritable one = new IntWritable(1);private Text word = new Text();public void map(Object key, Text value, Context context) throws IOException, InterruptedException {StringTokenizer itr = new StringTokenizer(value.toString());while (itr.hasMoreTokens()) {word.set(itr.nextToken());context.write(word, one);}}}public static class IntSumReducerextends Reducer<Text,IntWritable,Text,IntWritable> {private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {int sum = 0;for (IntWritable val : values) {sum += val.get();}result.set(sum);context.write(key, result);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = Job.getInstance(conf, "word count");job.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));System.exit(job.waitForCompletion(true) ? 0 : 1);}
}