1.spark 是什么？

spark 是一个开源的分布式计算引擎，专为大规模数据处理而设计。它提供了高级API，支持Java、Scala、Python 和R语言，能高效地执行批处理、流处理、图处理等多种工作负载。

spark 是apache 大数据生态的重要组成部分，主要用于替代hadoop 中的MapReduce 计算模型，在典型的大数据平台中，常见架构为：

HDFS（存储） + YARN（调度与资源管理） + Spark（计算引擎）

既然spark 主要是代替mapreduce，那就来看看两者的区别

维度	MapReduce	Spark
计算模型	仅支持 Map 和 Reduce 两个阶段	支持丰富的Transformation 和Action 操作（如 map、filter、join、reduceByKey 等API）
数据存储位置	中间结果必须写入磁盘	中间结果默认保留在内存中（可持久化到磁盘）
执行引擎	基于“Map → Shuffle → Reduce”的线性流程	基于DAG 的执行引擎，可优化任务调度
处理速度	较慢（频繁 I/O）	更快
容错机制	通过重跑失败任务实现	基于 RDD 的 血统（Lineage） 重建丢失数据
实时处理	不支持	支持（通过 Spark Streaming / Structured Streaming）
编程复杂度	较高（需编写 Mapper/Reducer 类）	更简洁（函数式编程风格）
适用场景	简单批处理、资源受限环境	复杂 ETL、流处理、图计算、机器学习、交互分析等

下面是同一个wordcount 程序分别用mapreduce 和spark 作执行引擎时需要编写的java 代码，都省略了import 语句。可以看到mapreduce 程序必须要编写mapper 和 reducer 类，并且只能处理简单批式任务，对于复杂任务是束手无策的。

mapreduce：

public class WordCountDriver {
    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("用法: WordCountDriver <输入路径> <输出路径>");
            System.exit(-1);
        }
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Word Count");
        job.setJarByClass(WordCountDriver.class);
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.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);
    }
}

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken().toLowerCase());
            context.write(word, one);
        }
    }
}

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();
    
    @Override
    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);
    }
}

spark：

public class SparkWordCount {
    public static void main(String[] args) {
        if (args.length != 2) {
            System.err.println("用法: WordCountDriver <输入路径> <输出路径>");
            System.exit(-1);
        }
        SparkConf conf = new SparkConf().setAppName("Word Count");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> textFile = sc.textFile(args[0]);
        JavaRDD<String> words = textFile.flatMap(line ->
                Arrays.asList(line.toLowerCase().split("\\s+")).iterator()
        ).filter(word -> !word.isEmpty());
        JavaPairRDD<String, Integer> wordPairs = words.mapToPair(word ->
                new Tuple2<>(word, 1)
        );
        JavaPairRDD<String, Integer> wordCounts = wordPairs.reduceByKey(
                (count1, count2) -> count1 + count2
        );
        wordCounts.saveAsTextFile(args[1]);
        sc.close();
    }
}

2. Spark 架构简介

spark 的主要架构包含三个部分：

• Driver：运行 main() 函数，构建 DAG，调度任务。
• Executor：在工作节点上执行具体任务，缓存数据。
• Cluster Manager：可以是 Standalone、YARN、Kubernetes 等。

Spark 架构图以及一个spark 任务执行的大致过程：

3.Spark 运行流程总览

1. 用户编写 Spark 应用程序
2. 启动 Driver 进程，创建 SparkContext
3. SparkContext 向集群管理器（Cluster Manager）申请资源
4. 集群启动 Executor 进程
5. 任务（Task）分发给 Executor 执行
6. Executor 执行计算并返回结果（或写入外部存储）
7. 应用结束，释放资源，Driver 退出，SparkContext 关闭，Executor释放

Spark 采用惰性求值（Lazy Evaluation） 机制：

• Transformation 操作（如 map, filter, join）只记录血统（Lineage），不立即执行。
• Action 操作（如 collect(), count(), saveAsTextFile()）才会真正触发计算。

当 Action 被调用时：

1. DAGScheduler 将 RDD 依赖关系划分为多个 Stage（以 Shuffle 为边界）。
2. 每个 Stage 包含多个 Task（每个分区对应一个 Task）。
3. TaskScheduler 将 Task 分发给可用的 Executor。

4.stage 划分机制

Stage 是 Spark 中一组可以并行执行且无需 Shuffle 的 Task 集合。
Spark 的 DAGScheduler 会根据 RDD 之间的依赖关系，将整个计算流程划分为多个 Stage，每个 Stage 内部只包含 窄依赖（Narrow Dependency） 操作。

stage 划分的核心依据：RDD 依赖类型

Spark 根据 RDD 之间的依赖关系 来决定是否切分 Stage：

窄依赖（不切分）：父 RDD 的一个分区最多被一个子 RDD 分区使用，即一对一或者多对一的关系（如 map, filter, union）

宽依赖（切分）：父 RDD 的一个分区会被子 RDD 的多个分区依赖(涉及到 shuffle)。即一对多的关系（如 groupByKey, reduceByKey, join, distinct）

public static void main(String[] args) {
        // Initialize Spark configuration and context
        SparkConf conf = new SparkConf().setAppName("RDD Word Count Example");
        JavaSparkContext sc = new JavaSparkContext(conf);

        // rdd = sc.textFile("hdfs://data.txt") - 4 partitions
        JavaRDD<String> rdd = sc.textFile("hdfs://data.txt", 4);

        // words = rdd.flatMap(lambda x: x.split()) - narrow transformation
        JavaRDD<String> words = rdd.flatMap(line -> 
            java.util.Arrays.asList(line.split("\\s+")).iterator()
        );

        // pairs = words.map(lambda w: (w, 1)) - narrow transformation
        JavaPairRDD<String, Integer> pairs = words.mapToPair(word -> 
            new Tuple2<>(word, 1)
        );

        // counts = pairs.reduceByKey(lambda a, b: a + b) - wide transformation (shuffle!)
        JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);

        // output = counts.filter(lambda x: x[1] > 10) - narrow transformation
        JavaPairRDD<String, Integer> output = counts.filter(tuple -> tuple._2() > 10);

        // output.saveAsTextFile("result") - action
        output.saveAsTextFile("result");

        // Close the Spark context
        sc.close();
    }

构建DAG：textFile → flatMap → map → reduceByKey → filter

反向遍历：DAGScheduler 从最终的 Action（saveAsTextFile）开始，反向扫描依赖链，直到遇到 reduceByKey 宽依赖：

• filter 和 saveAsTextFile 属于 ResultStage
• textFile → flatMap → map 属于 ShuffleMapStage

最终划分结果：

• Stage 0（ShuffleMapStage）：textFile → flatMap → map。输出中间结果（<word, 1>）到磁盘/内存，供 Shuffle 使用
• Stage 1（ResultStage）：reduceByKey → filter → saveAsTextFile。从 Shuffle 读取数据，聚合后过滤并写入文件