大连爱得科技网站建设公司怎么样,建设银行无锡分行招聘网站,服装如何做微商城网站,网站超市源码哪个好Spark 架构
Spark 是一个基于内存计算的大数据处理框架#xff0c;相比 Hadoop 的 MapReduce#xff0c;它能够提供 更高效的迭代计算 和 流式计算能力。Spark 采用 主从架构#xff08;Master-Slave#xff09;#xff0c;主要包括 Driver、Cluster Manager、Worker、Ex…Spark 架构
Spark 是一个基于内存计算的大数据处理框架相比 Hadoop 的 MapReduce它能够提供 更高效的迭代计算 和 流式计算能力。Spark 采用 主从架构Master-Slave主要包括 Driver、Cluster Manager、Worker、Executor 和 Task 等核心组件。
1. Spark 组件架构
1.1 核心组件
组件作用Driver驱动程序负责任务调度向 Cluster Manager 申请资源管理 SparkContext。Cluster Manager集群管理器负责资源调度如 Standalone、YARN、Mesos、Kubernetes。Worker工作节点运行在集群节点上管理 Executor 进程执行具体计算任务。Executor执行器由 Worker 启动执行 Spark 任务并存储中间计算数据。Task任务运行在 Executor 之上每个 Stage 被划分为多个 Task 并行执行。
1.2 Spark 架构示意图
------------------------------------------------------
| Driver |
| - 任务调度 |
| - 运行 SparkContext |
| - 将 Job 划分为多个 Stage |
------------------------------------------------------| 向集群管理器申请资源V
------------------------------------------------------
| Cluster Manager |
| - 资源调度 |
| - 可选Standalone / YARN / Mesos / Kubernetes |
------------------------------------------------------| 分配 Worker 节点V
---------------- ---------------- ----------------
| Worker 1 | | Worker 2 | | Worker 3 |
| - 启动 Executor | | - 启动 Executor | | - 启动 Executor |
| - 执行 Task | | - 执行 Task | | - 执行 Task |
---------------- ---------------- ----------------2. Spark 运行模式
Spark 可以运行在不同的集群管理器上
StandaloneSpark 自带的资源管理器简单易用适合小规模集群。YARNHadoop Yarn 集群适合 Hadoop 生态环境。MesosApache Mesos 集群适合多租户资源调度。KubernetesK8s 集群适用于云计算和容器化部署。
3. Spark 任务执行流程
Spark 任务的执行大致分为以下几个步骤
3.1 任务提交
Driver 进程启动 SparkContext并向 Cluster Manager 申请资源。Cluster Manager 分配 Worker 节点并在 Worker 上 启动 Executor。
3.2 Job 分解
Driver 将 Job 拆分为多个 Stage基于 DAG 计算。每个 Stage 由多个 Task 组成并被分配到不同的 Executor 运行。
3.3 Task 执行
Executor 执行 Task计算数据并存储中间结果RDD。Executor 向 Driver 汇报任务执行状态若失败则重新调度 Task。
3.4 结果返回
任务执行完成后Driver 收集最终计算结果存储到 HDFS、Kafka、MySQL 等。
4. Spark 计算模型
Spark 计算任务是基于 RDDResilient Distributed Dataset 和 DAG有向无环图 进行调度的。
4.1 RDD弹性分布式数据集
RDD 是 Spark 最核心的数据抽象提供
分区Partition数据被分成多个分区并行计算。容错性基于 Lineage血缘 记录转换关系支持自动恢复。惰性计算只有在 Action 触发时RDD 才会真正执行计算。
RDD 转换类型
Transformation转换如 map()、filter()、flatMap()不会立即执行。Action行动如 count()、collect()、saveAsTextFile()触发计算。
4.2 DAG有向无环图
Spark 任务会构建 DAGDAGScheduler将 RDD 之间的依赖关系转换为多个 Stage。每个 Stage 由 多个 Task 组成并行执行计算任务。
示例
val data sc.textFile(hdfs://input.txt) // RDD1
val words data.flatMap(_.split( )) // RDD2Transformation
val wordCount words.map((_, 1)) // RDD3Transformation
val result wordCount.reduceByKey(_ _) // RDD4Transformation
result.saveAsTextFile(hdfs://output.txt) // Action 触发计算Spark 内部执行过程
DAG 构建阶段 RDD1 - RDD2 - RDD3 - RDD4 Stage 划分阶段 flatMap() 和 map() 形成 Stage 1reduceByKey() 形成 Stage 2 Task 并行执行 每个 Stage 划分多个 Task并分发到 Executor 执行。
5. Spark 生态组件
Spark 具备丰富的生态系统适用于不同场景
组件作用Spark CoreRDD APIDAG 调度任务执行。Spark SQL运行 SQL 查询支持 DataFrame、Dataset API。Spark Streaming实时流处理支持 Kafka、Flume 等数据源。MLlib机器学习库支持 K-Means、决策树等算法。GraphX图计算引擎支持 PageRank、社区检测等。
6. Spark 与 Hadoop 对比
对比项SparkHadoopMapReduce计算模型RDD 内存计算磁盘读写速度高速适用于流计算慢适用于批处理容错机制RDD 通过 Lineage 恢复任务失败后重跑适用场景实时计算、流处理批处理、大规模数据存储
7. 适用场景
数据分析数据挖掘、数据清洗实时流计算结合 Kafka 实现流式数据处理机器学习推荐系统、分类预测图计算社交网络分析、PageRank
总结
Spark 采用 Driver Executor 的分布式架构基于 RDD 进行数据计算通过 DAG 调度任务并支持 SQL、流式计算、机器学习 等多种应用场景。相较于 HadoopSpark 计算更快适合 大数据分析、实时计算和 AI 训练。
Checkpoint
Spark 中的 Checkpoint 作用
Checkpoint检查点 主要用于 RDD 持久化和容错可以将 RDD 的数据存储到**持久化存储如 HDFS、S3**中以便在失败时快速恢复计算避免从头计算整个 DAG。
1. 为什么需要 Checkpoint
在 Spark 中RDD 具有血缘关系LineageSpark 通过血缘追踪来进行故障恢复。如果某个计算任务失败Spark 会重新从原始数据集按照血缘关系重新计算。 但是在以下情况下依赖血缘恢复可能导致 高额计算开销
RDD 计算链路太长如果 RDD 经过多次 Transformation失败后重新计算的开销会很大。Driver 内存溢出RDD 的血缘信息存储在 Driver 中过长的 Lineage 可能会导致 Driver 负担过重甚至 OOM。需要数据持久化某些情况下如流式计算需要持久化部分数据以便后续任务读取。
Checkpoint 可以 截断 RDD 血缘依赖将计算结果持久化避免重复计算提高容错能力。
2. Checkpoint 的作用
(1) 提高容错能力
在 RDD 发生丢失时不再依赖 Lineage 重新计算而是直接从持久化存储中加载数据提高恢复速度。
(2) 减少 DAG 依赖
通过 Checkpoint 截断 RDD 的血缘依赖避免 DAG 过长减少 Driver 负担。
(3) 持久化计算结果
适用于需要在不同任务中复用的 RDD如流式计算Spark Streaming中的状态数据。
3. Checkpoint vs Cache
CheckpointCache / Persist存储位置持久化到HDFS / S3 / 本地磁盘存储在Executor 的内存 / 磁盘数据存储方式持久化后会丢弃 RDD 血缘信息保留 RDD 血缘信息恢复方式任务失败后直接从 Checkpoint 读取任务失败后需要从头重新计算适用场景长计算链路 / 流式计算 / 容错短期数据复用 / 内存充足
Cache/Persist 适用于频繁访问数据但不能容错如果 Executor 挂掉数据会丢失需要重新计算。Checkpoint 适用于长计算 DAG 或需要持久化数据的场景但由于存储到 HDFS速度较慢。
4. Checkpoint 使用方式
(1) 开启 Checkpoint
在使用 Checkpoint 之前需要设置存储目录
import org.apache.spark.SparkContext
import org.apache.spark.SparkConfval conf new SparkConf().setAppName(CheckpointExample).setMaster(local[*])
val sc new SparkContext(conf)// 1. 设置 Checkpoint 存储路径
sc.setCheckpointDir(hdfs://namenode:9000/spark-checkpoint)// 2. 创建 RDD
val rdd sc.parallelize(Seq(1, 2, 3, 4, 5))// 3. 设置 Checkpoint
rdd.checkpoint()// 4. 触发计算
rdd.count()sc.setCheckpointDir(path) 设置 Checkpoint 目录必须是 HDFS、S3 或本地持久化存储。rdd.checkpoint() 标记 RDD 需要 Checkpoint。由于 Checkpoint 是惰性执行的必须在 Action如 count()、collect()时触发计算并存储。
(2) 与 Cache 结合使用
由于 Checkpoint 计算会重新执行整个 DAG可以先 cache()然后 checkpoint()避免重复计算
val rdd sc.textFile(hdfs://namenode:9000/data.txt).map(_.split( ))rdd.cache() // 缓存 RDD 避免重复计算
rdd.checkpoint() // 持久化数据rdd.count() // 触发计算cache() 先把数据缓存到内存避免在 checkpoint 时重复计算。
5. Checkpoint 在 Spark Streaming 中的应用
在 Spark Streaming 中Checkpoint 用于存储 Streaming 计算状态保证数据处理的容错性防止任务重启后状态丢失。
(1) 设置 Checkpoint 目录
import org.apache.spark.streaming.{Seconds, StreamingContext}// 创建 StreamingContext
val ssc new StreamingContext(sc, Seconds(5))// 设置 Checkpoint 目录
ssc.checkpoint(hdfs://namenode:9000/streaming-checkpoint)// 创建 DStream
val lines ssc.socketTextStream(localhost, 9999)
val wordCounts lines.flatMap(_.split( )).map(word (word, 1)).reduceByKey(_ _)// 启动流式计算
ssc.start()
ssc.awaitTermination()ssc.checkpoint(path) 设置 Checkpoint 目录用于存储流式计算的状态数据如窗口聚合数据。适用于 窗口操作window, updateStateByKey 场景。
6. 总结 Checkpoint 作用 持久化 RDD避免 DAG 过长导致计算性能下降。提高容错性避免 Executor 挂掉时重算整个 DAG。适用于 Streaming 计算存储流式数据状态。 使用方法 先 sc.setCheckpointDir() 设置目录。对 RDD 调用 checkpoint()。触发 Action如 count()来执行 checkpoint 计算。 Checkpoint vs Cache Cache/Persist 适用于临时缓存提高性能但不具备容错能力。Checkpoint 适用于长计算链路、流式计算保证容错但性能略慢。 最佳实践
长时间运行的任务如 Spark Streaming必须开启 Checkpoint。Checkpoint 和 Cache 结合使用避免重复计算导致性能下降。
并行度
Apache Spark 是一个分布式并行计算框架基于 RDD弹性分布式数据集 进行并行计算并利用集群资源提高计算效率。
Spark 的计算模型遵循 MapReduce 的思想但相比 HadoopSpark 采用 内存计算并且支持更加细粒度的任务调度和优化大大提升了计算性能。
Spark 的并行度parallelism 取决于以下几个因素
RDD 的分区数PartitionsExecutor 的数量CPU 核心数并行任务数Task 并发数
1. RDD 的分区数
在 Spark 中RDD 是由多个 分区Partitions 组成的每个分区可以在一个 Task 中独立计算因此分区数决定了并行度。
默认情况下 sc.textFile(path) 读取 HDFS 文件时分区数 HDFS block 数量通常是 128MB 一个 block。sc.parallelize(data, numSlices) 允许手动指定分区数 numSlices。
示例
val rdd sc.parallelize(Seq(1, 2, 3, 4, 5), 3) // 设置 3 个分区
println(rdd.partitions.length) // 输出: 3分区数越多并行度越高但过多的分区会导致 任务调度开销增加降低整体效率。
2. Executor 并行度
Executor 是 Spark 任务的执行单元每个 Executor 拥有多个 CPU 核心可同时运行多个 Task。
Executor 并行度计算方式 [ 并行度 Executors 数量 \times 每个 Executor 的 CPU 核心数 ]
例如
--num-executors 5 --executor-cores 4表示
5 个 Executors每个 Executor 4 核心最大并行 Task 数 5 × 4 20
3. 并行任务数Task 并发数
Spark 会按照RDD 分区数来决定 Task 数量并由集群的可用资源Executor 和 核心数来决定同时能运行的 Task 数量。
并行任务数计算公式 [ 并行任务数 min( RDD 分区数, 总 CPU 核心数 ) ] 例如
RDD 分区数 100Spark 资源 10 Executors每个 4 核心总可用核心数 10 × 4 40
并行度 min(100, 40) 40同时执行 40 个 Task
如何调整并行度
增加 RDD 分区数 rdd.repartition(n)增加或减少分区rdd.coalesce(n)减少分区避免数据洗牌 增加 Executor 核心数 --executor-cores N--num-executors M 增加 Task 并发 spark.default.parallelism全局默认并行度spark.sql.shuffle.partitionsSQL Shuffle 时的分区数 示例
val rdd sc.parallelize(Seq(1, 2, 3, 4, 5), 10) // 增加分区数提高并行度总结
影响因素说明影响并行度RDD 分区数任务并行度取决于分区数分区数越多并行度越高Executor 数量任务运行的执行节点数量Executors 越多并行度越高Executor 核心数每个 Executor 可并行运行的 Task 数核心数越多并行度越高Task 并发数Task 调度和 CPU 资源影响并发Task 数量受 CPU 资源限制 最佳实践
大数据计算时确保 RDD 分区数 ≥ 任务 CPU 核心数以充分利用计算资源。避免单个 Task 计算过长导致 CPU 资源利用率低下。Spark SQL 计算时适当调整 spark.sql.shuffle.partitions默认 200减少 Shuffle 代价。 结论 Spark 是 并行计算框架并行度主要由 RDD 分区数、Executor 数量、CPU 核心数、任务调度 共同决定合理调整参数可以优化计算性能。
