Spark基础-生产与应用
Spark运行过程
程序架构
- driver：负责控制整个spark应用程序的执行并维护集群状态。与集群管理器交互以获得物理资源并启动executor
- executor：执行由driver分配的任务，上报结果
- 集群管理器：负责维护一组运行spark程序的机器。spark目前支持三种集群管理器：
- 内置独立管理器
- Mesos
- Hadoop YARN
集群管理器由一个driver节点和若干工作节点构成，driver负责接收任务，并在worker节点启动任务。
执行模式
集群模式下，用户将预编译的Jar包提交给集群管理器，管理器在集群内某个工作节点启动driver进程。根据driver进程的请求，在集群工作节点上启动executor进程。
实线方框是spark的driver程序，虚线方框是spark的executor
客户端模式
和集群模式几乎一样，不同点在于spark的dirver保存在提交应用的客户端机器上。这意味着客户端机器维护driver进程，集群管理器维护executor进程。
本地模式
调试使用，在一台机器上运行程序，通过线程实现并行。
Spark程序生命周期（外部）
客户请求
通过本地客户端请求集群管理器，提交jar包，发送启动命令（红线），这一步只为spark driver请求资源。
资源管理器接受请求，将spark driver 启动在一个worker上
之后客户端退出，程序开始在集群上运行。
命令类似：
启动
spark driver开始执行代码，代码必须包含初始化spark集群的sparkSession。ss之后和集群管理器diriver通信，要求启动spark executor。
集群管理器会启动executor，并将executor位置等相关信息报告给spark driver
执行
节点间相互通信，executor按照driver的指示执行任务，上报结果。executor之间进行数据交换。
完成
应用完成后，driver会以成功或失败的状态退出。之后集群管理器关闭所有executor。
Spark程序生命周期（内部）
sparkSession
第一步就是创建SparkSession，所有spark任务执行都基于此
逻辑指令
根据编写的程序生成逻辑指令，之后再转换成具体的物理执行计划。
Spark作业
一个动作应该触发一个spark作业，调用动作总是会返回结果。每个作业（Job）被分解成一系列阶段（stage），其数量取决于进行多少次的shuffle。
上面这句话非常重要，这是spark任务拆解的核心逻辑。这里的动作指的是spark中的动作操作。简单理解：
动作操作决定job数量，shuffle次数决定stage数量
作业
每个动作操作对应一个job，因此一个spark应用可以包含多个job，在spark UI中会有体现。
阶段
spark中stage代表可以一起执行的任务组，用于在多台机器上执行相同的操作。spark将尽可能多的操作加入同一个阶段。引擎在shuffle之后将启动新的阶段。一次shuffle是一次对数据的物理重分区。
任务
spark中的阶段由若干任务（task）构成，每个任务对应一组数据和一组将在单个executor上进行的转换操作。任务个数和数据分区个数相等。即每一个数据分区都有一个任务来处理。
执行细节
流水线执行
如果不需要跨节点的数据移动，spark将一系列操作合并为一个单独的任务阶段，直接在内存中完成。
shuffle数据持久化
如果需要跨节点，则进行shuffle操作，此时会将数据持久化到磁盘。在已经执行了shuffle操作的数据上运行新的作业时，不会重新运行生产shuffle数据的任务。因为这些任务的运行结果已经被保存在磁盘了，只需要读取进行后续操作即可。这也是UI和日志中有些阶段会被标记为SKIPPED的原因。
Spark UI
选项卡
spark UI界面包含如下选项卡
- jobs对应spark作业
- stages对应各个阶段和相关任务
- storage包含当前在spark应用程序中缓存的信息和数据
- environment对应spark应用相关配置信息
- SQL对应我们提交的结构化API查询（SQL和DF）
- executors对应每个执行器的详细信息
用下面的例子详细解读：
启动一个spark应用来执行上述代码，包含读取csv，重分区为2，做一次筛选，做一次分组，然后计算count，再将结果输出到driver。
根据is_glass分组，所以最终is_glass为true，为false，为null分别对应一个count值，输出三行数据。
对应的SQL UI如下：
这里三个蓝色大框表示三个阶段，中间两个exchange小框就是进行了两次shuffle。可以看到，第一个阶段进行了csv文件的读取。然后执行repartiton，做了一次shuffle，将数据分为了2个分区。
第二阶段进行数据处理，做过滤，做分组聚合。这里还没有进行下一次的shuffle即数据汇总。因此是在两个分区上分别进行的。按照分组计算count的逻辑，两个分组各产生3条数据，总共6条数据。
第三阶段将来自两个分区的最后三行执行结果合并，并作为总查询结果输出。
三个阶段的任务个数可以在UI中查看
第一阶段有8个任务，第二阶段🈶2个，第三阶段有200个。
第一阶段的读取由8个任务并行执行，即将csv文件拆分成了8份进行。在集群上读取hdfs等可以拆分成更多份。
第二阶段有2个任务，因为我们人为指定了分区数为2
第三阶段有200个任务，因为默认的分区数是200，虽然如此，但真正能拿到数据的任务应该只有2个。
性能调优
性能调优遵循下面几个原则
API选择
尽量使用高级API，利用spark引擎的优化能力
序列化方式选择
开启kryo序列化
动态分配
根据工作负载动态调整应用占用的资源。spark.dynamicAllocation.enabled=true
数据存储
- 长期文件存储选择parquet格式
- 对表进行合理分区
- 分桶操作：分桶允许spark根据可能执行的连接操作或聚合操作对数据进行预先分区。可以提高稳定性和性能，避免数据倾斜。如果要根据某一列进行频繁的联结操作，则可以通过分桶来保证数据按照这一列进行了良好的分区
shuffle配置
shuffle中每个输出分区至少有几十兆字节的数据。保证每个工作节点不会太忙或太闲
shuffle操作代价高昂，但是如果通过平衡数据可以从整体上优化作业的执行，这个代价是值得的。一般shuffle尽可能少的数据。如果要减少分区数量，优先试试coalesce方法。
缓存
如果一个DF会被下游反复用到，可以选择显示指定将其缓存，这样下游无需重复执行生成DF的过程。
分区和分桶
分区(Partitioning):
- 分区是将数据集分割成较小的部分，每个分区可以在集群中的不同节点上进行处理。
- 分区可以通过对数据进行哈希分区、范围分区或列表分区来创建。
- 分区可以提高数据处理的并行性，同时减少数据传输的网络开销。
桶(Bucketing):
- 桶是将数据根据某个列的值进行分组，并将每个分组存储在独立的文件中。
- 桶可以根据数据的分布来优化查询性能，可以减少数据的扫描范围。
- 桶可以在创建表时指定，也可以在数据写入时进行动态分桶。