[docker/大数据]Spark快速入门

在这里插入图片描述

1. 概述

1.1 诞生背景

Spark官方文档：https://spark.apache.ac.cn/docs/latest/

Spark 由加州大学伯克利分校 AMP 实验室于 2009 年开发，2013 年成为 Apache 顶级项目，旨在解决 MapReduce 的三大核心问题：

功能局限：仅支持 Map 和 Reduce 两种操作，难以处理复杂计算
执行效率低：频繁的磁盘 I/O 操作导致性能瓶颈
生态割裂：需要与 Storm/Hive/HBase 等组件组合才能完成完整数据处理流程

1.2 核心特点

特点	说明	优势
极速处理	内存计算比 MapReduce 快 10-100 倍	实时分析能力
All-in-One	统一引擎支持多种计算范式	简化技术栈
易用性	支持 Scala/Java/Python/R/SQL	开发者友好
容错性	RDD 机制保障故障恢复	高可靠性

2. 核心概念

2.1 核心组件

Spark Core主要包含四大模块：Spark SQL（结构化数据处理）、Spark Streaming（流批次处理）、MLib（机器学习库）、GraphX（图计算模块）
在这里插入图片描述

案例：比如我们要基于用户行为日志进行分析，下面是各组件的分工：
在这里插入图片描述

Streaming：实时接收用户点击/购买数据流
SQL：将数据转为结构化表格，存储到数据仓库
MLlib：基于历史数据训练"看了又看"推荐模型
GraphX：分析用户间的设备/IP关联，识别刷单团伙
Core：底层支撑所有组件的分布式运行

2.2 RDD分布式数据集

RDD是Spark 最核心、最根本的数据抽象。如果说Spark是一个巨大的数据工厂，那么RDD就是工厂流水线上的一个个原材料零件、半成品。

2.2.1 特点

RDD（Resilient Distributed Dataset ）弹性分布式数据集特点：

特性	说明	优势
分布式	数据分区存储在集群节点	并行处理
弹性	支持高效故障恢复	容错性强
不可变	只读数据集	避免并发问题
延迟计算	操作按需执行	优化执行计划

1.分布式：

概念：你的数据量非常大，一台机器存不下、算不动。RDD 会把数据自动切割成很多份（Partitions/分区），分散存储在集群的多个机器上
类比：一本1000页的巨著，分给10个人一起读，每人读100页。这10个人就是一个“集群”，每个人手里的100页就是一个“分区”。
好处：并行计算，速度极快。

2.数据集：

概念：表示是数据的集合。RDD 里面可以存储任何类型的数据，比如数字、字符串、对象等。
- 类比：上面例子中，书里的文字内容就是“数据”。

3.弹性：

概念：有容错性，当有数据丢失时可以快速恢复数据，这也是 Spark 比 Hadoop MapReduce 快的关键原因，例如：集群中某个机器突然宕机了，它上面存储的那个数据分片（那100页书）丢了怎么办？传统方法需要从头重新计算。Spark的解决方案： RDD 记录了自己是如何从其他数据“计算”过来的（例如： “我是通过A文件经过filter操作再经过map操作得到的”）。这个记录叫做血统（Lineage）。
类比：读第3章需要先读第1章和第2章。如果某人手里的第2章丢了，我们不需要从头开始写书，只需要根据“依赖关系”（血统），让他去找有第1章的人，重新读一遍第1章，然后自己再推导出第2章即可。
好处：快速恢复数据，无需昂贵的数据复制备份。

2.2.2 RDD操作类型

RDD（Resilient Distributed Dataset ）又称弹性分布式数据集，操作主要包含两类：Transformation和Action。

①Transformation（转换，只做规划）：主要是规划；它会定义一个新的 RDD 是如何从现有的 RDD 计算过来的操作。它不会立即执行计算，只是在记录一个计算逻辑，而不是真正去算。
- map(): 对数据集中每个元素都执行一个函数。（例如：把每一行文字都变成大写）

filter(): 过滤掉不符合条件的元素。（例如：筛选出所有包含“错误”关键词的日志行）
groupByKey(): 对键值对数据按键进行分组。
reduceByKey(): 对每个键对应的值进行聚合计算。

类比：厨师拿到菜谱，菜谱上写着“1. 洗菜，2. 切菜，3. 炒菜”。此刻厨师只是在看菜谱，还没开始动手做。菜谱就是一系列的 Transformation。

②Action（动作，触发实际计算）：实际计算；触发实际计算，并返回结果给 Driver 程序或存储到外部系统的操作。

take(n): 取前n个元素。
saveAsTextFile(path): 将数据集保存到文件系统（如HDFS）。

类比：顾客说“老板，上菜！”。这时厨师才真正开始执行菜谱上的所有步骤（洗、切、炒）。这句“上菜”就是 Action。

在这里插入图片描述
Spark计算核心流程：定义Transformations -> 最后调用Action -> Spark生成执行计划 -> 分布式计算 -> 返回结果。

Q：为什么Spark需要惰性计算，即先规划，后计算？
A：Spark可以在看到所有“计划”（Transformation链）和一个最终“目标”（Action）后，对整个计算过程进行整体优化（比如合并一些操作），然后再执行，这样效率更高。

2.2.3 RDD依赖关系(宽窄依赖)

概念：一个 RDD 是由另一个或多个 RDD 通过 Transformation 计算得来的。这种“父子关系”就是依赖关系。

Spark 需要记录这种关系（即血统 Lineage），目的有两个：

容错：如果某个分区的数据丢了，可以根据血统关系重新计算。
任务调度：根据依赖类型来规划最有效的执行方式（这导致了 Stage 的划分）。

RDD的依赖关系分为两种：窄依赖（Narrow Dependency）和宽依赖（Wide Dependency / Shuffle Dependency）：

1. 窄依赖（窄关系）：

一对一：父 RDD 的每一个分区最多被子 RDD 的一个分区所使用。像一对一的护送，数据不需要在不同机器间移动（无需 Shuffle）。即：上游的RDD数据最多只会流到下游的一个RDD中，一对一的关系。
比如：map()、filter()、union() 这些

2. 宽依赖（宽关系）：

一对多：父RDD的一个分区被子RDD的多个分区所用。即：上游的RDD数据会流向下游多个RDD
比如：groupByKey()、reduceByKey()、sortByKey()这些

在这里插入图片描述

2.3 Spark 运行详解

2.3.1 运行架构

Spark应用程序以进程集合为单位在分布式集群上运行，通过driver程序的main方法创建的SparkContext对象与集群交互。
1、Spark通过SparkContext向Cluster manager(资源管理器)申请所需执行的资源（cpu、内存等）
2、Cluster manager分配应用程序执行需要的资源，在Worker节点上创建Executor
3、SparkContext 将程序代码（jar包或者python文件）和Task任务发送给Executor执行，并收集结果给Driver。
在这里插入图片描述

2.3.2 运行流程

在这里插入图片描述

2.3.3 功能介绍

1. Application[用户编写的应用程序]

指的是用户编写的Spark应用程序，包含了Driver功能代码和分布在集群中多个节点上运行的Executor代码。

Spark应用程序，由一个或多个作业JOB组成，如下图所示。
在这里插入图片描述

2. Driver：驱动程序

Spark中的Driver即运行上述Application的Main()函数并且创建SparkContext，其中创建SparkContext的目的是为了准备Spark应用程序的运行环境。在Spark中由SparkContext负责和ClusterManager通信，进行资源的申请、任务的分配和监控等；
当Executor部分运行完毕后，Driver负责将SparkContext关闭。通常SparkContext代表Driver，如下图所示。
在这里插入图片描述

3. Cluster Manager：资源管理器

指的是在集群上获取资源的外部服务，常用的有：Standalone，Spark原生的资源管理器，由Master负责资源的分配；Haddop Yarn，由Yarn中的ResearchManager负责资源的分配；Messos，由Messos中的Messos Master负责资源管理。

4. Executor：执行器

Application运行在Worker节点上的一个进程，该进程负责运行Task，并且负责将数据存在内存或者磁盘上，每个Application都有各自独立的一批Executor，如下图所示。
在这里插入图片描述

5. Worker：计算节点

集群中任何可以运行Application代码的节点，类似于Yarn中的NodeManager节点。在Standalone模式中指的就是通过Slave文件配置的Worker节点，在Spark on Yarn模式中指的就是NodeManager节点，在Spark on Messos模式中指的就是Messos Slave节点，如下图所示。
在这里插入图片描述

6. DAGScheduler：有向无环图调度器

基于DAG划分Stage 并以TaskSet的形式提交Stage给TaskScheduler；负责将作业拆分成不同阶段的具有依赖关系的多批任务；最重要的任务之一就是：计算作业和任务的依赖关系，制定调度逻辑。在SparkContext初始化的过程中被实例化，一个SparkContext对应创建一个DAGScheduler。
在这里插入图片描述

7. TaskScheduler：任务调度器

将Taskset提交给worker（集群）运行并回报结果；负责每个具体任务的实际物理调度。如图所示。
在这里插入图片描述

8. Job：作业

由一个或多个调度阶段所组成的一次计算作业；包含多个Task组成的并行计算，往往由Spark Action催生，一个JOB包含多个RDD及作用于相应RDD上的各种Operation。如图所示。
在这里插入图片描述

9. Stage：调度阶段

一个任务集对应的调度阶段；每个Job会被拆分很多组Task，每组任务被称为Stage，也可称TaskSet，一个作业分为多个阶段；Stage分成两种类型ShuffleMapStage、ResultStage。如图所示。
在这里插入图片描述
Application多个job多个Stage：Spark Application中可以因为不同的Action触发众多的job，一个Application中可以有很多的job，每个job是由一个或者多个Stage构成的，后面的Stage依赖于前面的Stage，也就是说只有前面依赖的Stage计算完毕后，后面的Stage才会运行。

划分依据：Stage划分的依据就是宽依赖，何时产生宽依赖，reduceByKey, groupByKey等算子，会导致宽依赖的产生。

核心算法：从后往前回溯，遇到窄依赖加入本stage，遇见宽依赖进行Stage切分。Spark内核会从触发Action操作的那个RDD开始从后往前推，首先会为最后一个RDD创建一个stage，然后继续倒推，如果发现对某个RDD是宽依赖，那么就会将宽依赖的那个RDD创建一个新的stage，那个RDD就是新的stage的最后一个RDD。然后依次类推，继续继续倒推，根据窄依赖或者宽依赖进行stage的划分，直到所有的RDD全部遍历完成为止。

将DAG划分为Stage剖析：如上图，从HDFS中读入数据生成3个不同的RDD，通过一系列transformation操作后再将计算结果保存回HDFS。可以看到这个DAG中只有join操作是一个宽依赖，Spark内核会以此为边界将其前后划分成不同的Stage. 同时我们可以注意到，在图中Stage2中，从map到union都是窄依赖，这两步操作可以形成一个流水线操作，通过map操作生成的partition可以不用等待整个RDD计算结束，而是继续进行union操作，这样大大提高了计算的效率。

10. TaskSet：任务集

由一组关联的，但相互之间没有Shuffle依赖关系的任务所组成的任务集。如图所示。
在这里插入图片描述
PS：
1）一个Stage创建一个TaskSet；
2）为Stage的每个Rdd分区创建一个Task,多个Task封装成TaskSet

11. Task：任务

被送到某个Executor上的工作任务；单个分区数据集上的最小处理流程单元（单个stage内部根据操作数据的分区数划分成多个task）。如图所示。
在这里插入图片描述
总体如图所示：

2.4 Spark 运行模式

Spark主要分为以下几种运行模式：

本地模式；
standalone模式；
spark on yarn 模式，又细分为两种子模式：yarn-client和yarn-cluster；
spark on mesos 模式
spark on cloud 模式

本文主要介绍前四种模式。

运行模式	资源管理者	核心特点	主要应用场景
本地模式	本地JVM	单机多线程模拟分布式计算，简单易用	开发、测试、学习
Standalone模式	Spark自带Master	Spark自带的独立集群模式，无需依赖其他资源管理系统	中小规模Spark专属集群
Spark on YARN	Hadoop YARN	利用Hadoop YARN进行资源调度，与Hadoop生态集成紧密，生产环境最常用	共享Hadoop集群的大规模生产环境
Spark on Mesos	Apache Mesos	更灵活的通用集群管理，支持细粒度调度（但细粒度模式已被弃用）	混合负载集群（如同时运行Spark、MPI等）

本地模式

概念：Spark不一定非要跑在hadoop集群，可以在本地，起多个线程的方式来指定。将Spark应用以多线程的方式直接运行在本地，一般都是为了方便调试，本地模式分三类

local：只启动一个executor
local[k]:启动k个executor
local[*]：启动跟cpu数目相同的executor

应用场景：一般用于开发测试。
执行流程：以 local[2] 为例

在IDE或Spark Shell中提交应用程序。
Spark会在本地启动一个JVM进程，这个进程既充当Driver（指挥者），又充当Executor（工作者）。
Driver会创建 SparkContext，并初始化调度器（如 TaskScheduler）。
SparkContext 会启动指定数量（此处为2个）的线程作为执行线程（Executor threads）。
所有的任务（Tasks）会在这2个线程中并行执行。
任务执行完毕后，结果返回到Driver，或写入本地文件系统。

Standalone模式

概念：它是一个主从式架构，包含Master（主节点）和Worker（从节点）。Master负责管理整个集群的资源，Worker负责在节点上启动Executor进程来执行具体任务。
应用场景：适用于中小规模的、专用的Spark集群。如果你不想依赖Hadoop YARN等其他资源管理系统，希望Spark独享集群资源，那么可以选择Standalone模式。

运行流程：

启动集群：事先在集群的每个节点上启动Spark的Master和Worker守护进程。
提交应用：用户通过spark-submit脚本或代码向Master提交应用程序。
资源申请：应用程序中的SparkContext向Master注册并申请资源（CPU和内存）。
启动Executor：Master根据Worker的心跳报告（汇报自身资源情况），在资源充足的Worker节点上启动Executor进程。
任务调度与执行：

Executor启动后，会向SparkContext反向注册。
SparkContext将应用程序代码（JAR包或Python文件）发送给Executor。
SparkContext根据程序中的RDD操作构建DAG图，并由DAGScheduler将其划分为Stage，再由TaskScheduler将每个Stage转化为一批Task，然后分发到各个Executor上执行。

结果与释放：Task执行完毕后，将结果返回给Driver或写入外部存储。所有任务完成后，SparkContext向Master注销，释放资源。

下面这个图也非常经典：

spark on yarn 模式

概念：Spark客户端直接连接Yarn。不需要额外构建Spark集群。分布式部署集群，资源和任务监控交给yarn管理，但是目前仅支持粗粒度资源分配方式。它根据Driver程序运行位置的不同分为cluster和client运行模式，cluster适合生产，driver运行在集群子节点，具有容错功能，client适合调试，dirver运行在客户端。

yarn-client模式：Driver程序运行在提交任务的客户端机器上。
yarn-cluster模式：Driver程序运行在YARN集群的某个NodeManager节点上（作为ApplicationMaster的一部分）。

应用场景：适用于公司已有Hadoop YARN集群，希望Spark与其他计算框架（如MapReduce）共享集群资源，统一管理的大规模生产环境。spark on yarn-client模式适合交互和调试（如：通过spark-shell），on yarn-Cluster模式更适合生产环境。

运行流程： (以 yarn-cluster 模式为例)：

提交应用：用户通过spark-submit脚本，指定–master yarn和–deploy-mode cluster，向YARN的ResourceManager提交应用程序。
启动ApplicationMaster：ResourceManager在某个NodeManager上分配一个Container，并在其中启动ApplicationMaster（AM）。注意：在cluster模式下，AM本身就包含了Spark Driver。
申请资源：AM（Driver）向ResourceManager申请运行Executor所需的资源。
启动Executor：ResourceManager分配Container后，AM与对应的NodeManager通信，在Container中启动Executor进程。
任务执行：Executor启动后，会向AM（Driver）注册。AM（Driver）中的SparkContext将Task分发给Executor执行。
完成与清理：应用运行完成后，AM会向ResourceManager注销并关闭自己，其占用的资源也随之释放。

PS：client模式与cluster模式区别

特性	YARN-Client 模式	YARN-Cluster 模式	核心提示
Driver 位置	客户端机器上	集群中的 ApplicationMaster 里	最根本的区别，决定了其他所有不同。
Application Master 角色	轻量，仅负责申请 Executor 资源	重量，就是 Driver 本身，负责全部调度	Cluster 模式的 AM 权力更大。
客户端要求	必须保持在线，直到应用结束	提交后即可断开	想关电脑就用 Cluster。
日志输出	直接输出到客户端控制台，便于调试	需要通过 yarn logs 命令或 Web UI 查看	调试用 Client，生产用 Cluster。
性能	网络通信可能跨网段，性能较差	Driver 在集群内，网络通信效率高	Cluster 模式性能更优。
应用场景	测试、调试、交互式查询	生产环境、长时间运行的任务	开发用 Client，上线用 Cluster。

yarn-client：

用于测试，因为driver运行在本地客户端，负责调度application，会与yarn集群产生超大量的网络通信。好处是直接执行时，本地可以看到所有的log，方便调试。
Application Master仅仅向YARN请求Executor，Client会和请求的Container通信来调度他们工作，也就是说Client不能离开。

yarn-cluster：

生产环境使用，因为driver运行在nodemanager上，缺点在于调试不方便，本地用spark-submit提价以后，看不到log，只能通过yarn application-logs application_id这种命令查看，很麻烦
Driver运行在AM(Application Master)中，它负责向YARN申请资源，并监督作业的运行状况。当用户提交了作业之后，就可以关掉Client，作业会继续在YARN上运行，因而YARN-Cluster模式不适合运行交互类型的作业；

spark on mesos 模式

概念：Spark可以作为Mesos框架上的一个应用程序运行，由Mesos来负责资源调度和分配。
应用场景：适用于需要运行多种类型计算框架（如同时运行Spark、MPI作业等）的混合负载集群，Mesos可以提供更灵活和通用的资源调度。
运行流程：(以粗粒度模式为例)

用户向Mesos Master提交Spark应用。
Mesos Master将资源offer发送给Spark的调度器（Driver）。
Spark Driver接受offer，并指示Mesos在提供资源的Slave节点上启动Executor。
Executor启动后，直接与Spark Driver通信，注册并申请Task。
Driver分配Task给Executor执行。
执行过程中，Executor直接向Driver汇报状态。

2.5 Stage 划分

划分原理

可以把 Spark 的任务执行想象成一个工厂的流水线：

DAG（有向无环图）：这是整个产品的生产流程图，由一系列步骤（RDD 转换操作）组成。
Stage（阶段）：流程图会被划分成几个大的生产阶段。划分的原则是：能否在流水线上不间断地完成一系列加工。
宽窄依赖：这是划分 Stage 的依据。
- 窄依赖：一个父零件只用来生产一个子零件。好比在同一个工位上对零件进行打磨、抛光。这些操作可以合并成一个 Stage，无需移动零件（无 Shuffle），效率极高。
- 宽依赖：需要把所有父零件打乱重组。好比把所有零件运到一个集中的装配区（Shuffle），按新的规则分组后才能进行下一步组装。这里必须划分成一个新的 Stage。

核心思想：Spark 的 DAGScheduler 会从最终结果倒推这个“流程图”，一旦遇到宽依赖（Shuffle），就画上一条分界线，形成一个新 Stage。每个 Stage 内部都是一连串的窄依赖，可以进行流水线优化（Pipeline），在内存中连续计算。

优化建议[调优]

优化方向	目标	具体方法	代码示例（不推荐 → 推荐）
减少 Shuffle	降低网络/磁盘IO	使用预聚合算子；使用广播Join	rdd.groupByKey().mapValues(sum) → rdd.reduceByKey(_ + _)
持久化 (Cache)	避免重复计算	对多次使用的RDD进行缓存	“val transformed = rdd.map(…)transformed.count(); transformed.reduce() → transformed.persist().count(); .reduce()”
并行度	充分利用资源	调整Shuffle后的分区数	使用 spark.sql.shuffle.partitions 参数
数据倾斜	避免长尾任务	对倾斜Key进行加盐处理	join(skewRdd) → addRandomPrefix(skewRdd).join(…).removePrefix()

PS：优化原理

减少与避免 Shuffle
为什么？ Shuffle 是分布式计算中最昂贵操作，涉及磁盘 I/O、网络 I/O、序列化/反序列化。减少一次 Shuffle，性能提升立竿见影。Stage 数量 ≈ Shuffle 次数。减少 Stage 数量本质上就是减少 Shuffle 次数。
持久化（缓存）的正确使用
为什么？如果一个 RDD 会被多个 Action 操作重用（例如一个循环里），默认情况下每次 Action 都会从头重新计算整个 RDD，极其浪费。
解决数据倾斜
为什么？如果某个 Task 的数据量远远超过其他 Task，导致绝大多数任务早就完成了，却在等那一个“慢哥”，会导致资源闲置。

建议：

写代码时时刻思考：“我这一步操作会不会引起 Shuffle？”
充分利用 Spark UI：提交任务后，一定要打开 Spark UI（通常是 http://:4040）。这是你最好的老师！在里面你可以看到：

整个执行的 DAG 可视化图，清晰看到有几个 Stage。
每个 Stage 的详情，有多少个 Task，花了多少时间。
Shuffle 读写的数据量，如果看到某个 Stage 写入了大量数据，就要警惕了。

从模仿开始：先记住 reduceByKey 比 groupByKey 好，broadcast join 比普通 join 好，在实际代码中先用起来。

3. 快速入门

搭建Spark环境

这里通过docker-compose搭建集群，如果没有的可以通过github下载。
docker compose github：https://github.com/docker/compose

查看本地是否有docker-compose

docker-compose version

在这里插入图片描述
2. 创建spark集群挂载目录

# 创建主目录
mkdir -p /Users/ziyi/docker-home/spark-cluster# 创建数据目录
mkdir -p /Users/ziyi/docker-home/spark-cluster/data# 创建 Spark Ivy 缓存目录
mkdir -p /Users/ziyi/docker-home/spark-cluster/spark-ivy# 设置权限
chmod -R 777 /Users/ziyi/docker-home/spark-cluster/

进入目录，创建docker-compose.yml文件

cd /Users/ziyi/docker-home/spark-cluster/
vi docker-compose.yml

docker-compose.yml：

    networks:spark-net:driver: bridgeipam:config:- subnet: 172.30.0.0/16 # 这里子网需要是docker本地没有被使用过的。可通过docker network ls + docker network inspect <network-id>查看services:# ================= Spark 集群配置 =================spark-master:image: bitnami/spark:3.5platform: linux/amd64container_name: spark-masterhostname: spark-masternetworks:spark-net:environment:- SPARK_MODE=master- SPARK_MASTER_HOST=spark-master- SPARK_MASTER_PORT=7077- SPARK_MASTER_WEBUI_PORT=8080- SPARK_USER=sparkvolumes:- /Users/ziyi/docker-home/spark-cluster/data:/tmp/data- /Users/ziyi/docker-home/spark-cluster/spark-ivy:/home/spark/.ivy2ports:- "8080:8080"- "7077:7077"- "6066:6066"spark-worker1:image: bitnami/spark:3.5platform: linux/amd64container_name: spark-worker1hostname: spark-worker1networks:spark-net:environment:- SPARK_MODE=worker- SPARK_WORKER_CORES=1- SPARK_WORKER_MEMORY=1g- SPARK_MASTER_URL=spark://spark-master:7077- SPARK_WORKER_PORT=8881- SPARK_WORKER_WEBUI_PORT=8081- SPARK_USER=sparkdepends_on:- spark-mastervolumes:- /Users/ziyi/docker-home/spark-cluster/data:/tmp/data- /Users/ziyi/docker-home/spark-cluster/spark-ivy:/home/spark/.ivy2ports:- "8081:8081"spark-worker2:image: bitnami/spark:3.5platform: linux/amd64container_name: spark-worker2hostname: spark-worker2networks:spark-net:environment:- SPARK_MODE=worker- SPARK_WORKER_CORES=1- SPARK_WORKER_MEMORY=1g- SPARK_MASTER_URL=spark://spark-master:7077- SPARK_WORKER_PORT=8882- SPARK_WORKER_WEBUI_PORT=8082- SPARK_USER=sparkdepends_on:- spark-mastervolumes:- /Users/ziyi/docker-home/spark-cluster/data:/tmp/data- /Users/ziyi/docker-home/spark-cluster/spark-ivy:/home/spark/.ivy2ports:- "8082:8082"spark-worker3:image: bitnami/spark:3.5platform: linux/amd64container_name: spark-worker3hostname: spark-worker3networks:spark-net:environment:- SPARK_MODE=worker- SPARK_WORKER_CORES=1- SPARK_WORKER_MEMORY=1g- SPARK_MASTER_URL=spark://spark-master:7077- SPARK_WORKER_PORT=8883- SPARK_WORKER_WEBUI_PORT=8083- SPARK_USER=sparkdepends_on:- spark-mastervolumes:- /Users/ziyi/docker-home/spark-cluster/data:/tmp/data- /Users/ziyi/docker-home/spark-cluster/spark-ivy:/home/spark/.ivy2ports:- "8083:8083"

启动集群

# 后台拉取镜像并启动集群
docker-compose up -d

在这里插入图片描述
5. 查看集群状态

docker-compose ps -a

在这里插入图片描述
可以看到所有容器状态都为Up，接下来可以访问UI页面：

Spark Master UI页面： http://localhost:8080
Spark 工作节点UI页面：
- Worker 1 Web UI: http://localhost:8081
- Worker 2 Web UI: http://localhost:8082
- Worker 3 Web UI: http://localhost:8083

在这里插入图片描述

Spark功能测试

方式一：使用spark-submit测试jar包

# 以 root 用户身份进入容器
docker exec -it -u root spark-master /bin/bash# 执行命令，提交运行spark测试任务
/opt/bitnami/spark/bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master spark://spark-master:7077 \--deploy-mode client \/opt/bitnami/spark/examples/jars/spark-examples*.jar 10

效果：
在这里插入图片描述

在Spark管理台上也可以看到作业运行状态：http://localhost:8080/
在这里插入图片描述

方式二：使用spark命令行验证

# 进入 spark-master 容器
docker exec -it -u root spark-master /bin/bash# 启动 Spark Shell
/opt/bitnami/spark/bin/spark-shell --master spark://spark-master:7077# 通过spark代码测试
scala> val rdd = spark.sparkContext.parallelize(1 to 1000)
scala> println(s"Count: ${rdd.count()}")
scala> val sum = rdd.reduce(_ + _)
scala> println(s"Sum: $sum")
scala> :quit

效果：
在这里插入图片描述

Spark Scala 案例：电商数据分析

官方API：https://spark.apache.org/docs/3.5.3/api/scala/org/apache/spark/index.html

案例文件创建

# 在主机上创建测试文件创建销售数据文件,因为我们集群搭建时将本地目录挂载到了容器内部的/tmp/data
# 所以容器/tmp/data/目录下可直接访问sales.csv文件
cat > /Users/ziyi/docker-home/spark-cluster/data/sales.csv << 'EOF'
id,product,category,price,quantity,date
1,ProductA,Electronics,100.0,2,2024-01-15
2,ProductB,Clothing,50.0,3,2024-01-16
3,ProductC,Electronics,200.0,1,2024-01-17
4,ProductD,Books,30.0,5,2024-01-18
5,ProductE,Clothing,80.0,2,2024-01-19
6,ProductF,Electronics,150.0,1,2024-01-20
7,ProductG,Books,25.0,4,2024-01-21
8,ProductH,Clothing,60.0,2,2024-01-22
EOF

进入容器，启动spark-shell

## 进入容器
docker exec -it -u root spark-master /bin/bash
## 进入spark-shell
/opt/bitnami/spark/bin/spark-shell --master spark://spark-master:7077

编写Scala语言，加载并分析数据

案例：电商销售数据分析。读取本地文件并分析

import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession// 1. 读取数据
// 通过file协议读取本地CSV 文件 （后续有hdfs可直接通过hdfs://读取hdfs数据分析）
val salesDF = spark.read.option("header", "true").option("inferSchema", "true").csv("file:///tmp/data/sales.csv")
salesDF.show()
salesDF.printSchema()
// 2. 添加计算列
val enrichedSales = salesDF.withColumn("revenue", $"price" * $"quantity")
enrichedSales.show()
// 3. 执行聚合分析
val analysisResult = enrichedSales.groupBy("category").agg(
count("*").as("transaction_count"),
sum("revenue").as("total_revenue"),
avg("revenue").as("avg_revenue"))// 4. 排序并显示最终结果
analysisResult.orderBy(desc("total_revenue")).show()