04.spark

1 介绍

1.1 定义

01.Spark概念
    Apache Spark是统一的大数据处理引擎
    支持批处理和流处理
    提供高级API和丰富的库
    运行在集群上的分布式计算框架
    由加州大学伯克利分校AMPLab开发

02.设计目标
    提供快速的大数据处理能力
    统一批处理和流处理
    易于使用的高级API
    支持多种编程语言
    与Hadoop生态系统集成

03.核心特点
    内存计算,速度快
    通用的数据处理平台
    支持SQL,流处理,机器学习,图计算
    容错性强
    可扩展性好

1.2 概念：核心组件

01.Spark Core
    Spark的基础执行引擎
    提供RDD抽象
    任务调度和内存管理
    故障恢复机制
    所有其他组件的基础

02.Spark SQL
    结构化数据处理
    支持SQL查询
    DataFrame和Dataset API
    多种数据源支持
    Catalyst优化器

03.Spark Streaming
    实时流处理
    微批处理模型
    DStream抽象
    与批处理统一API
    Structured Streaming

04.Spark MLlib
    机器学习库
    分类,回归,聚类算法
    特征工程
    模型评估
    Pipeline API

05.Spark GraphX
    图计算框架
    图抽象和操作
    图算法库
    Pregel API
    与RDD集成

1.3 优缺点

01.优点
    a.高性能
        内存计算比Hadoop MapReduce快100倍
        磁盘计算快10倍
        DAG执行引擎
        优化的执行计划
    b.易用性
        高级API简化开发
        支持Scala,Java,Python,R
        交互式Shell
        丰富的算子
    c.通用性
        统一的数据处理平台
        批处理,流处理,机器学习,图计算
        避免多个系统集成
        降低学习成本
    d.兼容性
        与Hadoop生态集成
        支持HDFS,HBase,Hive
        可以运行在YARN,Mesos,Kubernetes
        读写多种数据源

02.缺点
    a.内存消耗大
        内存计算需要大量内存
        数据量超过内存时性能下降
        需要合理配置内存
    b.学习曲线
        概念较多需要理解
        调优需要经验
        分布式计算复杂
    c.小文件问题
        大量小文件影响性能
        需要合并小文件
        增加管理复杂度
    d.实时性
        微批处理有延迟
        不适合毫秒级实时
        Structured Streaming改进

1.4 使用场景

01.批处理
    大规模数据ETL
    日志分析
    数据仓库
    报表生成
    离线计算

02.实时处理
    实时数据分析
    流式ETL
    实时监控
    事件处理
    实时推荐

03.机器学习
    大规模机器学习
    特征工程
    模型训练
    模型评估
    在线预测

04.图计算
    社交网络分析
    推荐系统
    网页排名
    路径查找
    社区发现

05.交互式查询
    即席查询
    数据探索
    BI分析
    数据可视化

1.5 Spark vs Hadoop

01.计算模型
    a.Spark
        基于内存的迭代计算
        DAG执行引擎
        RDD抽象
        支持迭代算法
    b.Hadoop MapReduce
        基于磁盘的批处理
        Map和Reduce两阶段
        中间结果写磁盘
        迭代效率低
    c.性能对比
        Spark内存计算快100倍
        Spark磁盘计算快10倍
        Spark适合迭代算法
        MapReduce适合简单批处理

02.易用性
    a.Spark
        高级API:RDD,DataFrame,Dataset
        支持Scala,Java,Python,R
        交互式Shell
        丰富的算子
        代码简洁
    b.Hadoop MapReduce
        底层API
        只支持Java
        代码冗长
        开发效率低
    c.对比
        Spark开发效率高
        Spark代码量少
        Spark学习曲线平缓

03.生态系统
    a.Spark
        Spark SQL,Streaming,MLlib,GraphX
        统一平台
        组件集成好
        避免数据移动
    b.Hadoop
        Hive,Pig,HBase,Mahout
        组件独立
        需要集成
        数据移动开销大
    c.对比
        Spark生态更统一
        Hadoop生态更成熟
        Spark可以运行在Hadoop上

04.适用场景
    a.Spark适合
        迭代算法
        交互式查询
        实时处理
        机器学习
        图计算
    b.Hadoop适合
        简单批处理
        超大数据集
        对内存要求低
        稳定性要求高
    c.结合使用
        Spark运行在YARN上
        读写HDFS数据
        利用Hadoop资源管理
        发挥各自优势

1.6 运行架构

01.集群架构
    a.Driver
        运行main函数
        创建SparkContext
        划分任务
        调度任务到Executor
        收集结果
    b.Executor
        运行在Worker节点
        执行Task
        存储RDD数据
        向Driver汇报状态
        多个Executor并行执行
    c.Cluster Manager
        资源管理器
        分配资源给应用
        监控节点状态
        支持Standalone,YARN,Mesos,Kubernetes

02.任务执行流程
    a.Job提交
        用户提交Spark应用
        Driver创建SparkContext
        连接Cluster Manager
        申请资源
    b.任务划分
        根据RDD依赖关系构建DAG
        DAGScheduler划分Stage
        每个Stage包含多个Task
        TaskScheduler调度Task
    c.任务执行
        Task分配到Executor
        Executor执行Task
        读取数据,计算,写结果
        汇报执行状态
    d.结果收集
        Executor返回结果给Driver
        Driver聚合结果
        输出最终结果

03.调度机制
    a.DAGScheduler
        高层调度器
        根据RDD依赖划分Stage
        宽依赖划分Stage边界
        窄依赖在同一Stage
        提交Stage给TaskScheduler
    b.TaskScheduler
        底层调度器
        将Task分配到Executor
        管理Task执行
        处理Task失败重试
        数据本地性优化
    c.调度策略
        FIFO:先进先出
        FAIR:公平调度
        可配置调度池
        支持任务优先级

04.数据本地性
    a.本地性级别
        PROCESS_LOCAL:数据在同一JVM
        NODE_LOCAL:数据在同一节点
        RACK_LOCAL:数据在同一机架
        ANY:数据在任意位置
    b.调度优化
        优先调度本地性高的Task
        减少网络传输
        提高性能
        本地性等待时间可配置

1.7 部署模式

01.Standalone模式
    a.特点
        Spark自带的集群管理器
        简单易用
        适合小规模集群
        独立部署
    b.架构
        Master:资源管理和任务调度
        Worker:执行任务
        Driver和Executor
    c.部署
        启动Master
        启动Worker
        提交应用到Master
        适合开发测试

02.YARN模式
    a.特点
        运行在Hadoop YARN上
        与Hadoop集成
        共享集群资源
        生产环境常用
    b.Client模式
        Driver运行在客户端
        适合交互式应用
        客户端不能关闭
    c.Cluster模式
        Driver运行在YARN容器中
        适合生产环境
        客户端可以关闭
        更稳定

03.Mesos模式
    a.特点
        运行在Apache Mesos上
        细粒度资源共享
        支持多种框架
        动态资源分配
    b.粗粒度模式
        预先分配资源
        资源独占
        性能好
    c.细粒度模式
        动态分配资源
        资源利用率高
        适合多租户

04.Kubernetes模式
    a.特点
        运行在Kubernetes上
        容器化部署
        云原生
        动态扩缩容
    b.部署方式
        Driver和Executor运行在Pod中
        使用Kubernetes资源管理
        支持动态分配
        适合云环境
    c.优势
        容器隔离
        自动恢复
        滚动更新
        与云平台集成

05.Local模式
    a.特点
        单机运行
        用于开发调试
        不需要集群
        简单快速
    b.使用
        spark-shell --master local
        local[N]:N个线程
        local[*]:所有CPU核心
    c.应用场景
        学习Spark
        开发调试
        单元测试
        小数据集处理

1.8 内存管理

01.内存模型
    a.堆内内存
        Executor JVM堆内存
        存储RDD数据
        执行任务
        垃圾回收管理
    b.堆外内存
        直接内存
        避免GC开销
        序列化数据存储
        需要手动管理
    c.内存分区
        Storage Memory:缓存RDD
        Execution Memory:任务执行
        User Memory:用户数据结构
        Reserved Memory:系统保留

02.统一内存管理
    a.动态分配
        Storage和Execution共享内存
        动态调整边界
        提高内存利用率
        Spark 1.6+默认
    b.内存借用
        Execution可以借用Storage内存
        Storage可以借用Execution内存
        有空闲时借用
        需要时归还
    c.配置参数
        spark.memory.fraction:统一内存占比,默认0.6
        spark.memory.storageFraction:Storage占比,默认0.5
        动态调整

03.Storage内存
    a.RDD缓存
        cache()和persist()
        存储RDD分区数据
        加速迭代计算
        LRU淘汰策略
    b.存储级别
        MEMORY_ONLY:仅内存
        MEMORY_AND_DISK:内存+磁盘
        MEMORY_ONLY_SER:序列化存储
        DISK_ONLY:仅磁盘
        OFF_HEAP:堆外内存
    c.淘汰机制
        LRU算法
        内存不足时淘汰
        淘汰到磁盘或重算
        保证任务执行

04.Execution内存
    a.用途
        Shuffle过程
        Join操作
        排序操作
        聚合计算
    b.Shuffle内存
        Map端输出缓冲
        Reduce端拉取数据
        排序缓冲区
        溢写磁盘
    c.内存不足处理
        溢写到磁盘
        增加Shuffle分区数
        调整内存配置
        优化算法

05.内存调优
    a.合理配置内存
        根据数据量配置
        预留足够内存
        避免OOM
        监控内存使用
    b.选择存储级别
        根据数据特点选择
        频繁使用用MEMORY
        大数据用MEMORY_AND_DISK
        序列化节省空间
    c.减少内存使用
        过滤不需要的数据
        使用高效数据结构
        避免数据倾斜
        合理分区
    d.GC调优
        使用G1 GC
        调整GC参数
        减少Full GC
        监控GC日志

2 核心组件

2.1 汇总：5个

00.总结
    a.Spark Core
        基础执行引擎
        RDD抽象
        任务调度
        内存管理
        所有组件的基础
    b.Spark SQL
        结构化数据处理
        SQL查询
        DataFrame和Dataset
        多数据源支持
        Catalyst优化器
    c.Spark Streaming
        实时流处理
        微批处理
        DStream抽象
        与批处理统一
        Structured Streaming
    d.Spark MLlib
        机器学习库
        分类回归聚类
        特征工程
        Pipeline API
        分布式算法
    e.Spark GraphX
        图计算框架
        图抽象
        图算法
        Pregel API
        与RDD集成

2.2 Spark Core

01.核心功能
    a.RDD抽象
        弹性分布式数据集
        不可变分布式集合
        容错机制
        惰性计算
        分区存储
    b.任务调度
        DAG调度
        Stage划分
        Task分配
        数据本地性
        失败重试
    c.内存管理
        统一内存管理
        Storage和Execution内存
        缓存管理
        溢写磁盘
        GC优化

02.编程模型
    a.转换操作
        map,filter,flatMap
        返回新RDD
        惰性执行
        构建计算链
    b.行动操作
        collect,count,reduce
        触发计算
        返回结果
        提交Job
    c.持久化
        cache()和persist()
        存储中间结果
        加速迭代
        多种存储级别

03.应用场景
    批处理计算
    迭代算法
    交互式查询
    其他组件基础

2.3 Spark SQL

01.核心功能
    a.SQL查询
        标准SQL支持
        HiveQL兼容
        复杂查询
        窗口函数
        子查询
    b.DataFrame
        分布式数据表
        Schema信息
        优化执行
        多语言API
        与RDD互转
    c.Dataset
        类型安全
        编译时检查
        面向对象
        函数式API
        Scala和Java支持

02.数据源
    a.内置数据源
        Parquet,ORC,JSON
        CSV,Text
        JDBC
        Hive表
        Avro
    b.外部数据源
        Cassandra
        MongoDB
        Elasticsearch
        Kafka
        自定义数据源
    c.统一API
        read和write API
        options配置
        schema推断
        分区读写

03.Catalyst优化器
    a.逻辑优化
        谓词下推
        列裁剪
        常量折叠
        表达式简化
    b.物理优化
        代码生成
        向量化执行
        Join策略选择
        分区裁剪
    c.CBO优化
        基于成本优化
        统计信息
        选择最优计划
        动态优化

04.应用场景
    结构化数据处理
    ETL任务
    数据仓库
    BI分析
    与Hive集成

2.4 Spark Streaming

01.核心概念
    a.DStream
        离散化流
        RDD序列
        时间窗口
        微批处理
        容错机制
    b.输入源
        Kafka,Flume,Kinesis
        Socket,文件
        自定义Receiver
        Direct API
    c.输出操作
        print,saveAsTextFiles
        foreachRDD
        自定义输出
        exactly-once语义

02.窗口操作
    a.滑动窗口
        window()
        窗口长度
        滑动间隔
        时间窗口聚合
    b.状态操作
        updateStateByKey
        mapWithState
        维护状态
        增量计算
    c.Join操作
        Stream-Stream Join
        Stream-Static Join
        窗口Join
        时间对齐

03.Structured Streaming
    a.统一API
        与批处理相同API
        DataFrame/Dataset
        SQL查询
        易于使用
    b.事件时间
        水位线
        延迟数据处理
        窗口聚合
        精确一次
    c.输出模式
        Append:追加
        Complete:完整
        Update:更新
        灵活输出

04.应用场景
    实时数据分析
    流式ETL
    实时监控
    事件处理
    实时推荐

2.5 Spark MLlib

01.机器学习算法
    a.分类算法
        逻辑回归
        决策树
        随机森林
        梯度提升树
        朴素贝叶斯
        支持向量机
    b.回归算法
        线性回归
        广义线性回归
        决策树回归
        随机森林回归
        梯度提升树回归
    c.聚类算法
        K-Means
        高斯混合模型
        LDA
        二分K-Means
        层次聚类

02.特征工程
    a.特征提取
        TF-IDF
        Word2Vec
        CountVectorizer
        特征哈希
    b.特征转换
        标准化
        归一化
        PCA降维
        多项式展开
        独热编码
    c.特征选择
        卡方检验
        相关性分析
        方差阈值
        递归特征消除

03.Pipeline API
    a.Transformer
        特征转换
        模型预测
        链式调用
        可复用
    b.Estimator
        算法训练
        fit方法
        生成Model
        参数学习
    c.Pipeline
        工作流
        多阶段
        参数调优
        交叉验证
        模型持久化

04.模型评估
    a.分类评估
        准确率
        精确率和召回率
        F1分数
        ROC和AUC
        混淆矩阵
    b.回归评估
        RMSE
        MSE
        MAE
        R²
    c.聚类评估
        轮廓系数
        Davies-Bouldin指数
        Calinski-Harabasz指数

05.应用场景
    推荐系统
    文本分类
    图像识别
    异常检测
    预测分析

2.6 Spark GraphX

01.图抽象
    a.Graph类
        顶点RDD
        边RDD
        三元组视图
        不可变图
    b.属性图
        顶点属性
        边属性
        类型安全
        灵活建模
    c.图构建
        Graph()构造
        GraphLoader加载
        边列表构建
        邻接表构建

02.图操作
    a.结构操作
        reverse:反转边
        subgraph:子图
        mask:掩码
        groupEdges:合并边
    b.Join操作
        joinVertices:顶点Join
        outerJoinVertices:外Join
        mapVertices:映射顶点
        mapEdges:映射边
    c.聚合操作
        aggregateMessages
        collectNeighbors
        degrees:度数
        triangleCount:三角形计数

03.图算法
    a.PageRank
        网页排名
        迭代计算
        收敛判断
        个性化PageRank
    b.连通分量
        强连通分量
        弱连通分量
        标签传播
        社区发现
    c.最短路径
        单源最短路径
        全对最短路径
        Dijkstra算法
        Bellman-Ford算法
    d.其他算法
        三角形计数
        SVD++推荐
        标签传播
        图着色

04.Pregel API
    a.BSP模型
        批量同步并行
        超步迭代
        消息传递
        收敛检测
    b.编程模型
        顶点程序
        发送消息
        聚合消息
        更新状态
    c.应用
        自定义图算法
        迭代计算
        分布式图处理

05.应用场景
    社交网络分析
    推荐系统
    网页排名
    路径查找
    社区发现
    知识图谱

3 Spark Core

3.1 RDD概念

01.RDD定义
    a.弹性分布式数据集
        Resilient Distributed Dataset
        不可变的分布式对象集合
        分区存储在集群节点上
        并行操作
        容错机制
    b.五大特性
        分区列表:数据分片
        计算函数:作用于每个分区
        依赖关系:RDD之间的依赖
        分区器:键值RDD的分区策略
        数据位置:计算的最优位置
    c.抽象特点
        不可变性
        分区性
        容错性
        惰性计算
        可缓存

02.RDD优势
    a.内存计算
        数据缓存在内存
        避免磁盘IO
        加速迭代计算
        性能提升显著
    b.容错机制
        血统信息Lineage
        丢失分区可重算
        无需数据复制
        高效容错
    c.粗粒度操作
        对所有元素应用相同操作
        适合批处理
        高效并行
        简化编程

03.RDD局限性
    a.不适合细粒度更新
        无法高效更新单个元素
        需要重新计算整个分区
        不适合在线事务
    b.不适合异步迭代
        迭代之间有依赖
        需要等待上一轮完成
        图计算效率较低

3.2 RDD创建

01.并行化集合
    a.parallelize方法
        val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5))
        从本地集合创建
        指定分区数
        适合小数据集
    b.makeRDD方法
        val rdd = sc.makeRDD(Seq(1, 2, 3, 4, 5))
        等价于parallelize
        可以指定位置偏好
    c.分区控制
        sc.parallelize(data, 10)
        第二个参数指定分区数
        默认根据集群配置

02.读取外部数据
    a.文本文件
        val rdd = sc.textFile("hdfs://path/to/file")
        读取文本文件
        每行作为一个元素
        支持通配符
        支持压缩文件
    b.SequenceFile
        val rdd = sc.sequenceFile[K, V]("path")
        读取Hadoop SequenceFile
        键值对RDD
        高效二进制格式
    c.对象文件
        val rdd = sc.objectFile[T]("path")
        读取序列化对象
        保持类型信息
        Spark专用格式
    d.其他格式
        sc.wholeTextFiles("path")  // 整个文件
        sc.binaryFiles("path")  // 二进制文件
        sc.hadoopFile()  // Hadoop InputFormat
        sc.newAPIHadoopFile()  // 新API

03.从其他RDD转换
    a.转换操作
        val rdd2 = rdd1.map(_ * 2)
        通过转换创建新RDD
        惰性执行
        构建计算链
    b.依赖关系
        新RDD依赖于父RDD
        记录血统信息
        用于容错

3.3 RDD转换操作

01.基本转换
    a.map映射
        rdd.map(x => x * 2)
        一对一转换
        应用函数到每个元素
        返回新RDD
    b.filter过滤
        rdd.filter(x => x > 0)
        筛选满足条件的元素
        返回新RDD
        可能改变分区大小
    c.flatMap扁平化
        rdd.flatMap(x => x.split(" "))
        一对多转换
        扁平化结果
        处理嵌套结构
    d.mapPartitions
        rdd.mapPartitions(iter => iter.map(_ * 2))
        作用于整个分区
        减少函数调用开销
        可以初始化连接

02.键值对转换
    a.mapValues
        pairRDD.mapValues(v => v * 2)
        只转换值
        保持键不变
        保持分区器
    b.flatMapValues
        pairRDD.flatMapValues(v => v.split(","))
        值的扁平化映射
        保持键
        一对多
    c.keys和values
        pairRDD.keys  // 所有键
        pairRDD.values  // 所有值
        提取键或值

03.集合操作
    a.union合并
        rdd1.union(rdd2)
        合并两个RDD
        不去重
        分区数相加
    b.intersection交集
        rdd1.intersection(rdd2)
        两个RDD的交集
        去重
        需要Shuffle
    c.subtract差集
        rdd1.subtract(rdd2)
        rdd1中不在rdd2中的元素
        去重
        需要Shuffle
    d.distinct去重
        rdd.distinct()
        去除重复元素
        需要Shuffle
        可以指定分区数

04.分区操作
    a.repartition重分区
        rdd.repartition(10)
        改变分区数
        Shuffle操作
        可增可减
    b.coalesce减少分区
        rdd.coalesce(5)
        减少分区数
        避免Shuffle(默认)
        优化性能
    c.partitionBy
        pairRDD.partitionBy(new HashPartitioner(10))
        自定义分区器
        键值对RDD
        影响后续操作

3.4 RDD行动操作

01.聚合操作
    a.reduce归约
        rdd.reduce((a, b) => a + b)
        聚合所有元素
        返回单个值
        需要结合律
    b.fold折叠
        rdd.fold(0)((a, b) => a + b)
        带初始值的归约
        每个分区使用初始值
        结果再聚合
    c.aggregate聚合
        rdd.aggregate(0)(
          (acc, v) => acc + v,
          (acc1, acc2) => acc1 + acc2
        )
        两个函数:局部和全局
        灵活聚合
        可以改变类型

02.收集操作
    a.collect收集
        val array = rdd.collect()
        收集所有元素到Driver
        返回数组
        小心OOM
    b.take获取
        val array = rdd.take(10)
        获取前N个元素
        返回数组
        不需要计算所有分区
    c.first第一个
        val elem = rdd.first()
        获取第一个元素
        等价于take(1).head
    d.top前N个
        val array = rdd.top(10)
        获取最大的N个元素
        需要排序
        返回数组

03.统计操作
    a.count计数
        val n = rdd.count()
        元素个数
        触发计算
        返回Long
    b.countByKey
        val map = pairRDD.countByKey()
        每个键的元素个数
        返回Map
        结果收集到Driver
    c.countByValue
        val map = rdd.countByValue()
        每个值的出现次数
        返回Map
        需要Shuffle

04.保存操作
    a.saveAsTextFile
        rdd.saveAsTextFile("hdfs://path")
        保存为文本文件
        每个元素一行
        支持压缩
    b.saveAsSequenceFile
        pairRDD.saveAsSequenceFile("path")
        保存为SequenceFile
        键值对RDD
        二进制格式
    c.saveAsObjectFile
        rdd.saveAsObjectFile("path")
        保存为对象文件
        序列化存储
        保持类型

05.遍历操作
    a.foreach
        rdd.foreach(println)
        对每个元素执行操作
        在Executor上执行
        无返回值
    b.foreachPartition
        rdd.foreachPartition(iter => {
          // 处理整个分区
        })
        对每个分区执行操作
        可以初始化连接
        批量处理

3.5 RDD持久化

01.缓存级别
    a.MEMORY_ONLY
        rdd.cache()或rdd.persist()
        仅内存存储
        反序列化对象
        最快但占内存
        默认级别
    b.MEMORY_AND_DISK
        rdd.persist(StorageLevel.MEMORY_AND_DISK)
        内存+磁盘
        内存不足溢写磁盘
        平衡性能和可靠性
    c.MEMORY_ONLY_SER
        rdd.persist(StorageLevel.MEMORY_ONLY_SER)
        序列化存储
        节省内存
        需要反序列化开销
    d.DISK_ONLY
        rdd.persist(StorageLevel.DISK_ONLY)
        仅磁盘存储
        节省内存
        性能较低
    e.OFF_HEAP
        rdd.persist(StorageLevel.OFF_HEAP)
        堆外内存
        避免GC
        需要配置

02.持久化策略
    a.何时缓存
        多次使用的RDD
        计算代价高的RDD
        迭代算法
        交互式查询
    b.何时不缓存
        只使用一次
        内存不足
        数据量太大
        简单计算
    c.缓存位置
        优先内存
        内存不足用磁盘
        考虑序列化
        根据场景选择

03.缓存管理
    a.unpersist释放
        rdd.unpersist()
        释放缓存
        手动管理
        及时释放
    b.LRU淘汰
        内存不足自动淘汰
        最近最少使用
        自动管理
        无需干预
    c.监控缓存
        Spark UI查看
        Storage标签
        缓存大小和位置
        淘汰情况

04.Checkpoint
    a.设置检查点
        sc.setCheckpointDir("hdfs://path")
        rdd.checkpoint()
        持久化到可靠存储
        截断血统
    b.与缓存区别
        checkpoint写可靠存储
        cache写内存或本地磁盘
        checkpoint截断依赖
        cache保留依赖
    c.使用场景
        血统链太长
        宽依赖太多
        容错开销大
        迭代算法

3.6 RDD分区

01.分区概念
    a.分区定义
        RDD的数据分片
        并行计算单元
        分布在不同节点
        独立处理
    b.分区数量
        影响并行度
        影响任务数
        影响性能
        可配置
    c.分区器
        HashPartitioner:哈希分区
        RangePartitioner:范围分区
        自定义分区器
        键值对RDD

02.分区操作
    a.查看分区
        rdd.partitions.length  // 分区数
        rdd.partitions  // 分区数组
        rdd.getNumPartitions  // 分区数
    b.重新分区
        rdd.repartition(10)  // 增加或减少
        rdd.coalesce(5)  // 减少分区
        pairRDD.partitionBy(partitioner)  // 自定义
    c.分区函数
        rdd.mapPartitions(iter => ...)
        rdd.mapPartitionsWithIndex((idx, iter) => ...)
        rdd.foreachPartition(iter => ...)
        作用于分区

03.分区器
    a.HashPartitioner
        new HashPartitioner(numPartitions)
        key.hashCode % numPartitions
        默认分区器
        均匀分布
    b.RangePartitioner
        new RangePartitioner(numPartitions, rdd)
        按键范围分区
        采样确定边界
        有序数据
    c.自定义分区器
        class CustomPartitioner(numParts: Int) extends Partitioner {
          def numPartitions = numParts
          def getPartition(key: Any) = {
            // 自定义分区逻辑
          }
        }
        实现Partitioner
        自定义逻辑

04.分区优化
    a.合理设置分区数
        一般为CPU核心数的2-3倍
        数据量大增加分区
        避免数据倾斜
        平衡并行度和开销
    b.保持分区器
        某些操作保持分区器
        mapValues,flatMapValues
        避免不必要的Shuffle
        提高性能
    c.数据本地性
        分区靠近数据
        减少网络传输
        提高性能
        调度优化

3.7 宽窄依赖

01.窄依赖
    a.定义
        父RDD的每个分区最多被一个子RDD分区使用
        一对一或多对一
        不需要Shuffle
        可以流水线执行
    b.典型操作
        map,filter,flatMap
        mapPartitions
        union
        mapValues
    c.特点
        高效执行
        失败恢复快
        可以合并到一个Stage
        数据本地性好

02.宽依赖
    a.定义
        父RDD的一个分区被多个子RDD分区使用
        一对多
        需要Shuffle
        Stage边界
    b.典型操作
        groupByKey,reduceByKey
        join,cogroup
        distinct,repartition
        sortByKey
    c.特点
        需要Shuffle
        性能开销大
        失败恢复慢
        划分Stage

03.依赖影响
    a.Stage划分
        宽依赖划分Stage边界
        窄依赖在同一Stage
        影响并行度
        影响调度
    b.容错恢复
        窄依赖恢复快
        宽依赖需要重算多个分区
        影响容错性能
    c.性能优化
        减少宽依赖
        合并窄依赖
        优化Shuffle
        提高性能

04.Shuffle优化
    a.减少Shuffle
        使用reduceByKey代替groupByKey
        使用mapPartitions批处理
        广播小表
        预分区
    b.调整参数
        spark.shuffle.file.buffer
        spark.reducer.maxSizeInFlight
        spark.shuffle.sort.bypassMergeThreshold
        优化性能
    c.监控Shuffle
        Spark UI查看
        Shuffle读写量
        Shuffle时间
        定位瓶颈

3.8 容错机制

01.血统机制
    a.Lineage
        记录RDD的依赖关系
        形成有向无环图DAG
        用于容错恢复
        无需数据复制
    b.重新计算
        分区丢失时重算
        根据血统信息
        只重算丢失的分区
        高效容错
    c.血统截断
        checkpoint截断血统
        避免血统过长
        减少恢复开销
        写可靠存储

02.任务重试
    a.失败重试
        Task失败自动重试
        默认重试4次
        spark.task.maxFailures配置
        不同节点重试
    b.推测执行
        检测慢任务
        启动备份任务
        先完成的结果有效
        spark.speculation.enabled
    c.黑名单机制
        记录失败节点
        避免重复调度
        提高成功率
        自动恢复

03.数据容错
    a.RDD不可变
        创建后不可修改
        避免并发问题
        简化容错
        线程安全
    b.分区复制
        某些存储级别支持复制
        MEMORY_ONLY_2
        MEMORY_AND_DISK_2
        提高可靠性
    c.Checkpoint
        持久化到可靠存储
        HDFS等
        截断血统
        容错保障

04.容错最佳实践
    a.合理使用缓存
        频繁使用的RDD缓存
        减少重算开销
        选择合适级别
        及时释放
    b.设置Checkpoint
        血统链长时使用
        迭代算法
        关键中间结果
        定期检查点
    c.监控失败
        查看失败任务
        分析失败原因
        优化代码和配置
        提高稳定性

4 Spark SQL

4.1 DataFrame

01.DataFrame概念
    a.定义
        分布式数据表
        带Schema的RDD
        类似关系型数据库表
        列式存储优化
    b.特点
        Schema信息
        优化执行计划
        多语言API
        与RDD互转
        类型安全
    c.优势
        Catalyst优化
        代码生成
        列式存储
        性能更好

02.创建DataFrame
    a.从RDD创建
        val df = rdd.toDF("col1", "col2")
        val df = spark.createDataFrame(rdd, schema)
        指定列名或Schema
    b.从数据源创建
        val df = spark.read.json("path")
        val df = spark.read.parquet("path")
        val df = spark.read.csv("path")
        多种格式支持
    c.从Hive表创建
        val df = spark.table("tableName")
        val df = spark.sql("SELECT * FROM table")
        直接查询Hive表

03.DataFrame操作
    a.选择列
        df.select("col1", "col2")
        df.select($"col1", $"col2" + 1)
        df.selectExpr("col1", "col2 * 2 as doubled")
        灵活选择
    b.过滤行
        df.filter($"age" > 18)
        df.where("age > 18")
        条件过滤
    c.聚合操作
        df.groupBy("dept").count()
        df.groupBy("dept").agg(avg("salary"), max("age"))
        分组聚合
    d.排序
        df.orderBy($"age".desc)
        df.sort("name", "age")
        排序操作

4.2 Dataset

01.Dataset概念
    a.定义
        类型安全的DataFrame
        编译时类型检查
        面向对象API
        Scala和Java支持
    b.与DataFrame关系
        DataFrame是Dataset[Row]
        Dataset是强类型
        可以互相转换
        统一API
    c.优势
        类型安全
        编译时检查
        面向对象
        性能优化

02.创建Dataset
    a.从序列创建
        val ds = Seq(1, 2, 3).toDS()
        val ds = spark.createDataset(Seq(Person("Alice", 25)))
        类型推断
    b.从DataFrame转换
        val ds = df.as[Person]
        指定类型
        编译时检查
    c.从RDD转换
        val ds = rdd.toDS()
        自动推断类型

03.Dataset操作
    a.类型安全操作
        ds.map(p => p.age * 2)
        ds.filter(p => p.age > 18)
        ds.flatMap(p => p.name.split(" "))
        编译时检查
    b.SQL风格操作
        ds.select($"name", $"age")
        ds.where($"age" > 18)
        ds.groupBy("dept").count()
        与DataFrame相同
    c.转换
        ds.toDF()  // 转DataFrame
        ds.rdd  // 转RDD
        灵活转换

4.3 SQL查询

01.SQL基础
    a.注册临时表
        df.createOrReplaceTempView("people")
        注册为临时表
        可以SQL查询
        会话级别
    b.执行SQL
        val result = spark.sql("SELECT * FROM people WHERE age > 18")
        标准SQL语法
        返回DataFrame
    c.全局临时表
        df.createGlobalTempView("people")
        全局可见
        跨会话
        global_temp数据库

02.SQL功能
    a.基本查询
        SELECT,WHERE,GROUP BY
        ORDER BY,LIMIT
        JOIN,UNION
        子查询
    b.聚合函数
        COUNT,SUM,AVG
        MAX,MIN
        DISTINCT
        窗口函数
    c.复杂查询
        CASE WHEN
        HAVING
        嵌套查询
        CTE

03.HiveQL支持
    a.Hive兼容
        支持HiveQL语法
        Hive函数
        Hive表
        Hive分区
    b.Hive集成
        读写Hive表
        Hive元数据
        Hive SerDe
        Hive UDF

4.4 数据源

01.内置数据源
    a.Parquet
        spark.read.parquet("path")
        列式存储
        压缩高效
        Schema自动推断
        默认格式
    b.JSON
        spark.read.json("path")
        半结构化数据
        Schema推断
        嵌套结构支持
    c.CSV
        spark.read.option("header", "true").csv("path")
        文本格式
        分隔符可配置
        Schema推断
    d.ORC
        spark.read.orc("path")
        列式存储
        Hive优化格式
        高压缩比

02.数据库数据源
    a.JDBC
        spark.read.format("jdbc")
          .option("url", "jdbc:mysql://host:port/db")
          .option("dbtable", "table")
          .option("user", "user")
          .option("password", "pwd")
          .load()
        关系型数据库
        分区读取
        并行加载
    b.Hive表
        spark.table("tableName")
        spark.sql("SELECT * FROM table")
        直接访问
        元数据集成

03.外部数据源
    a.Kafka
        spark.read.format("kafka")
          .option("kafka.bootstrap.servers", "host:port")
          .option("subscribe", "topic")
          .load()
        流数据源
        批处理模式
    b.Cassandra
        需要spark-cassandra-connector
        分布式NoSQL
        并行读取
    c.MongoDB
        需要mongo-spark-connector
        文档数据库
        灵活Schema
    d.Elasticsearch
        需要elasticsearch-spark
        搜索引擎
        全文检索

04.读写选项
    a.通用选项
        .option("key", "value")
        .options(Map("k1" -> "v1", "k2" -> "v2"))
        配置参数
    b.保存模式
        .mode("overwrite")  // 覆盖
        .mode("append")  // 追加
        .mode("ignore")  // 忽略
        .mode("error")  // 错误(默认)
    c.分区写入
        .partitionBy("col1", "col2")
        分区存储
        提高查询性能

4.5 UDF函数

01.注册UDF
    a.匿名函数
        spark.udf.register("addOne", (x: Int) => x + 1)
        注册函数
        SQL中使用
    b.命名函数
        def addOne(x: Int): Int = x + 1
        spark.udf.register("addOne", addOne _)
        复用函数
    c.DataFrame API
        val addOneUDF = udf((x: Int) => x + 1)
        df.select(addOneUDF($"col"))
        不需要注册

02.UDF类型
    a.标量UDF
        输入单个值
        返回单个值
        最常用
    b.聚合UDF
        输入多个值
        返回单个值
        自定义聚合
    c.表生成UDF
        输入单个值
        返回多行
        UDTF

03.UDF最佳实践
    a.性能考虑
        UDF无法优化
        尽量使用内置函数
        避免复杂逻辑
        考虑性能影响
    b.类型安全
        指定输入输出类型
        避免运行时错误
        使用强类型
    c.序列化
        UDF需要序列化
        避免引用不可序列化对象
        使用闭包

4.6 窗口函数

01.窗口函数概念
    a.定义
        在一组相关行上计算
        不改变行数
        每行都有结果
        类似SQL窗口函数
    b.窗口规范
        partitionBy:分区
        orderBy:排序
        rowsBetween:行范围
        rangeBetween:值范围
    c.应用场景
        排名
        累计计算
        移动平均
        同比环比

02.排名函数
    a.row_number
        import org.apache.spark.sql.expressions.Window
        val w = Window.partitionBy("dept").orderBy($"salary".desc)
        df.select($"*", row_number().over(w).as("rank"))
        连续排名
        1,2,3,4
    b.rank
        df.select($"*", rank().over(w).as("rank"))
        并列排名
        1,2,2,4
    c.dense_rank
        df.select($"*", dense_rank().over(w).as("rank"))
        密集排名
        1,2,2,3

03.聚合函数
    a.累计和
        val w = Window.partitionBy("dept").orderBy("date")
          .rowsBetween(Window.unboundedPreceding, Window.currentRow)
        df.select($"*", sum("amount").over(w).as("cumsum"))
        累计计算
    b.移动平均
        val w = Window.orderBy("date").rowsBetween(-2, 0)
        df.select($"*", avg("value").over(w).as("ma3"))
        3日移动平均
    c.lead和lag
        df.select($"*",
          lag("value", 1).over(w).as("prev"),
          lead("value", 1).over(w).as("next"))
        前后值

04.窗口范围
    a.rowsBetween
        基于行数
        Window.unboundedPreceding:开始
        Window.currentRow:当前行
        Window.unboundedFollowing:结束
        数值:偏移量
    b.rangeBetween
        基于值范围
        适合时间窗口
        值的范围

4.7 性能优化

01.Catalyst优化器
    a.逻辑优化
        谓词下推
        列裁剪
        常量折叠
        表达式简化
        自动优化
    b.物理优化
        Join策略选择
        Broadcast Join
        Sort Merge Join
        代码生成
    c.CBO优化
        基于成本优化
        统计信息
        选择最优计划
        ANALYZE TABLE收集统计

02.缓存优化
    a.缓存DataFrame
        df.cache()
        df.persist(StorageLevel.MEMORY_AND_DISK)
        频繁使用的数据
        加速查询
    b.缓存表
        spark.catalog.cacheTable("tableName")
        缓存整个表
        SQL查询加速
    c.清除缓存
        df.unpersist()
        spark.catalog.uncacheTable("tableName")
        释放内存

03.分区优化
    a.数据分区
        df.repartition(10)
        df.repartition($"col")
        合理分区数
        避免数据倾斜
    b.分区裁剪
        读取分区表
        WHERE条件过滤分区
        减少数据扫描
        提高性能
    c.动态分区
        写入时动态创建分区
        .partitionBy("col")
        组织数据

04.Join优化
    a.Broadcast Join
        小表广播
        避免Shuffle
        spark.sql.autoBroadcastJoinThreshold
        默认10MB
    b.Sort Merge Join
        大表Join
        排序后合并
        适合大数据集
    c.Shuffle Hash Join
        中等数据集
        哈希Join
        性能适中

05.其他优化
    a.列式存储
        使用Parquet或ORC
        压缩高效
        列裁剪
        谓词下推
    b.向量化执行
        spark.sql.parquet.enableVectorizedReader
        批量处理
        SIMD指令
        提高性能
    c.代码生成
        Whole-stage代码生成
        减少虚函数调用
        提高CPU效率
        自动启用

5 Spark Streaming

5.1 DStream

01.DStream概念
    a.定义
        离散化流Discretized Stream
        连续的RDD序列
        每个RDD包含一个时间间隔的数据
        微批处理模型
    b.特点
        时间窗口
        容错机制
        与RDD API一致
        批流统一
    c.批次间隔
        设置批次大小
        影响延迟和吞吐
        一般500ms-10s
        根据场景调整

02.创建DStream
    a.基本输入源
        val ssc = new StreamingContext(sc, Seconds(1))
        val lines = ssc.socketTextStream("localhost", 9999)
        Socket流
        文件流
    b.高级输入源
        KafkaUtils.createDirectStream
        FlumeUtils.createStream
        KinesisUtils.createStream
        需要额外依赖
    c.自定义Receiver
        继承Receiver类
        实现onStart和onStop
        自定义数据接收逻辑

03.DStream操作
    a.转换操作
        lines.map(_.split(" "))
        lines.filter(_.length > 0)
        lines.flatMap(_.split(" "))
        类似RDD操作
    b.输出操作
        lines.print()
        lines.saveAsTextFiles("prefix")
        lines.foreachRDD(rdd => ...)
        触发计算
    c.启动和停止
        ssc.start()
        ssc.awaitTermination()
        ssc.stop()
        流式应用生命周期

5.2 输入源

01.基本输入源
    a.文件流
        ssc.textFileStream("hdfs://path")
        监控目录
        新文件自动处理
        支持通配符
    b.Socket流
        ssc.socketTextStream("host", port)
        TCP Socket
        测试使用
        不推荐生产
    c.RDD队列
        ssc.queueStream(queueOfRDDs)
        测试DStream
        模拟流数据

02.Kafka输入源
    a.Receiver方式
        KafkaUtils.createStream(ssc, zkQuorum, group, topics)
        使用Receiver接收
        WAL保证可靠性
        可能重复消费
    b.Direct方式
        KafkaUtils.createDirectStream(ssc, kafkaParams, topics)
        直接读取Kafka
        exactly-once语义
        手动管理offset
        推荐使用
    c.配置
        bootstrap.servers
        group.id
        auto.offset.reset
        enable.auto.commit

03.其他输入源
    a.Flume
        FlumeUtils.createStream
        Avro Sink
        Push或Pull模式
    b.Kinesis
        KinesisUtils.createStream
        AWS流服务
        区域配置
    c.自定义
        继承Receiver
        实现receive逻辑
        错误处理

5.3 转换操作

01.无状态转换
    a.map
        dstream.map(x => x * 2)
        一对一转换
        应用于每个RDD
    b.flatMap
        dstream.flatMap(_.split(" "))
        一对多转换
        扁平化结果
    c.filter
        dstream.filter(_.length > 0)
        过滤元素
        返回新DStream
    d.repartition
        dstream.repartition(10)
        改变分区数
        并行度调整

02.有状态转换
    a.updateStateByKey
        def updateFunc(values: Seq[Int], state: Option[Int]): Option[Int] = {
          Some(state.getOrElse(0) + values.sum)
        }
        dstream.updateStateByKey(updateFunc)
        维护全局状态
        所有历史数据
    b.mapWithState
        val stateSpec = StateSpec.function(mappingFunc)
        dstream.mapWithState(stateSpec)
        更高效的状态管理
        超时机制
        推荐使用

03.窗口操作
    a.window
        dstream.window(Seconds(30), Seconds(10))
        滑动窗口
        窗口长度和滑动间隔
        批次间隔的倍数
    b.countByWindow
        dstream.countByWindow(Seconds(30), Seconds(10))
        窗口内元素计数
        滑动计数
    c.reduceByWindow
        dstream.reduceByWindow(_ + _, Seconds(30), Seconds(10))
        窗口内归约
        增量计算
    d.reduceByKeyAndWindow
        pairDStream.reduceByKeyAndWindow(_ + _, Seconds(30), Seconds(10))
        按键窗口归约
        高效计算

5.4 输出操作

01.基本输出
    a.print
        dstream.print()
        打印前10个元素
        调试使用
        触发计算
    b.saveAsTextFiles
        dstream.saveAsTextFiles("prefix")
        保存为文本文件
        每个批次一个目录
        时间戳后缀
    c.saveAsObjectFiles
        dstream.saveAsObjectFiles("prefix")
        序列化保存
        保持类型信息

02.foreachRDD
    a.基本用法
        dstream.foreachRDD { rdd =>
          rdd.foreach(record => {
            // 处理每条记录
          })
        }
        自定义输出
        灵活处理
    b.连接管理
        dstream.foreachRDD { rdd =>
          rdd.foreachPartition { partition =>
            val connection = createConnection()
            partition.foreach(record => connection.send(record))
            connection.close()
          }
        }
        每个分区一个连接
        批量发送
        高效输出
    c.连接池
        使用连接池
        复用连接
        提高性能
        避免频繁创建

03.输出到数据库
    a.JDBC输出
        dstream.foreachRDD { rdd =>
          rdd.foreachPartition { partition =>
            val conn = DriverManager.getConnection(url)
            partition.foreach { record =>
              val stmt = conn.prepareStatement(sql)
              stmt.setString(1, record)
              stmt.executeUpdate()
            }
            conn.close()
          }
        }
        批量插入
        事务处理
    b.HBase输出
        使用HBase API
        批量Put
        异步写入
    c.Elasticsearch输出
        使用ES客户端
        批量索引
        错误处理

04.输出到Kafka
    a.Producer方式
        dstream.foreachRDD { rdd =>
          rdd.foreachPartition { partition =>
            val producer = new KafkaProducer[String, String](props)
            partition.foreach { record =>
              producer.send(new ProducerRecord(topic, record))
            }
            producer.close()
          }
        }
        发送到Kafka
        异步发送
    b.连接池
        使用Producer池
        复用Producer
        提高性能

5.5 窗口操作

01.窗口概念
    a.窗口长度
        窗口包含的时间范围
        必须是批次间隔的倍数
        如批次1s,窗口可以是30s
    b.滑动间隔
        窗口移动的间隔
        必须是批次间隔的倍数
        如滑动间隔10s
    c.窗口计算
        每个滑动间隔计算一次
        包含窗口长度内的所有数据
        可能重叠

02.窗口转换
    a.window
        dstream.window(Seconds(30), Seconds(10))
        返回窗口DStream
        包含窗口内所有RDD
        可以应用任何RDD操作
    b.countByWindow
        dstream.countByWindow(Seconds(30), Seconds(10))
        窗口内元素计数
        返回单值DStream
    c.reduceByWindow
        dstream.reduceByWindow(
          (a, b) => a + b,
          Seconds(30),
          Seconds(10)
        )
        窗口内归约
        聚合操作

03.增量窗口操作
    a.reduceByWindow增量
        dstream.reduceByWindow(
          (a, b) => a + b,  // 加入窗口
          (a, b) => a - b,  // 离开窗口
          Seconds(30),
          Seconds(10)
        )
        增量计算
        高效
        可逆操作
    b.reduceByKeyAndWindow
        pairDStream.reduceByKeyAndWindow(
          (a, b) => a + b,
          (a, b) => a - b,
          Seconds(30),
          Seconds(10)
        )
        按键增量窗口
        更高效
    c.性能优化
        使用增量操作
        避免重复计算
        checkpoint支持
        提高性能

04.窗口应用
    a.实时统计
        最近N分钟的访问量
        滑动窗口计数
        实时监控
    b.趋势分析
        移动平均
        趋势检测
        异常发现
    c.会话分析
        用户会话窗口
        超时检测
        行为分析

5.6 状态管理

01.updateStateByKey
    a.基本用法
        def updateFunc(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
          val newCount = newValues.sum + runningCount.getOrElse(0)
          Some(newCount)
        }
        pairDStream.updateStateByKey(updateFunc)
        维护全局状态
        累计计算
    b.特点
        维护所有键的状态
        内存占用大
        性能较低
        需要checkpoint
    c.应用场景
        累计统计
        全局计数
        用户状态跟踪

02.mapWithState
    a.基本用法
        def mappingFunc(key: String, value: Option[Int], state: State[Int]): Option[(String, Int)] = {
          val sum = value.getOrElse(0) + state.getOption.getOrElse(0)
          state.update(sum)
          Some((key, sum))
        }
        val stateSpec = StateSpec.function(mappingFunc)
        pairDStream.mapWithState(stateSpec)
        更高效的状态管理
    b.特点
        只处理有更新的键
        支持超时
        性能更好
        推荐使用
    c.超时管理
        stateSpec.timeout(Seconds(60))
        自动清理过期状态
        节省内存
        避免状态膨胀

03.Checkpoint
    a.设置checkpoint
        ssc.checkpoint("hdfs://path")
        必须设置
        状态恢复
        容错机制
    b.checkpoint间隔
        dstream.checkpoint(Seconds(10))
        控制checkpoint频率
        平衡性能和可靠性
    c.从checkpoint恢复
        StreamingContext.getOrCreate(checkpointDir, createFunc)
        自动恢复
        Driver失败恢复
        状态保持

04.状态最佳实践
    a.选择合适的状态管理
        简单场景用updateStateByKey
        复杂场景用mapWithState
        考虑性能和内存
    b.状态清理
        设置超时
        定期清理
        避免内存溢出
    c.checkpoint优化
        合理设置间隔
        使用可靠存储
        监控checkpoint性能

5.7 Structured Streaming

01.基本概念
    a.定义
        基于Spark SQL的流处理
        统一批流API
        DataFrame/Dataset
        声明式查询
    b.优势
        与批处理相同API
        Catalyst优化
        exactly-once语义
        事件时间处理
    c.编程模型
        输入表
        查询
        结果表
        输出模式

02.创建流
    a.文件源
        val df = spark.readStream
          .format("csv")
          .option("header", "true")
          .schema(schema)
          .load("path")
        监控目录
        自动处理新文件
    b.Kafka源
        val df = spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "host:port")
          .option("subscribe", "topic")
          .load()
        消费Kafka
        exactly-once
    c.Socket源
        val df = spark.readStream
          .format("socket")
          .option("host", "localhost")
          .option("port", 9999)
          .load()
        测试使用

03.流式查询
    a.DataFrame操作
        df.select("col1", "col2")
        df.where("col1 > 10")
        df.groupBy("col1").count()
        与批处理相同
    b.聚合查询
        df.groupBy("window", "col1").count()
        窗口聚合
        事件时间
    c.Join操作
        stream1.join(stream2, "key")
        流流Join
        流批Join
        水位线对齐

04.输出模式
    a.Append模式
        .outputMode("append")
        只输出新增行
        不能有聚合(除窗口)
        默认模式
    b.Complete模式
        .outputMode("complete")
        输出完整结果表
        适合聚合查询
        内存占用大
    c.Update模式
        .outputMode("update")
        只输出更新的行
        适合聚合查询
        平衡性能和完整性

05.输出Sink
    a.Console Sink
        query.writeStream
          .format("console")
          .start()
        控制台输出
        调试使用
    b.File Sink
        query.writeStream
          .format("parquet")
          .option("path", "path")
          .option("checkpointLocation", "checkpoint")
          .start()
        文件输出
        支持多种格式
    c.Kafka Sink
        query.writeStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "host:port")
          .option("topic", "topic")
          .start()
        输出到Kafka
    d.ForeachBatch
        query.writeStream
          .foreachBatch { (batchDF, batchId) =>
            // 处理每个批次
          }
          .start()
        自定义输出
        批处理逻辑

06.事件时间和水位线
    a.事件时间
        df.withColumn("timestamp", $"event_time")
        使用事件发生时间
        而非处理时间
    b.水位线
        df.withWatermark("timestamp", "10 minutes")
        处理延迟数据
        超过水位线丢弃
        窗口触发
    c.延迟数据处理
        水位线内的延迟数据
        更新窗口结果
        超过水位线丢弃
        平衡延迟和性能

6 RDD API详解

6.1 创建RDD

01.并行化集合
    a.parallelize方法
        val data = Seq(1, 2, 3, 4, 5)
        val rdd = sc.parallelize(data)
        从本地集合创建
        适合小数据集
    b.指定分区数
        val rdd = sc.parallelize(data, 10)
        第二个参数指定分区数
        控制并行度
    c.makeRDD方法
        val rdd = sc.makeRDD(Seq((1, "a"), (2, "b")))
        等价于parallelize
        可以指定位置偏好

02.读取外部数据
    a.文本文件
        val rdd = sc.textFile("hdfs://path/to/file")
        val rdd = sc.textFile("file:///local/path")
        val rdd = sc.textFile("s3://bucket/path")
        读取文本文件
        支持多种文件系统
        支持通配符
    b.多个文件
        val rdd = sc.textFile("hdfs://path/*.txt")
        val rdd = sc.wholeTextFiles("hdfs://path/")
        批量读取
        wholeTextFiles返回(文件名,内容)
    c.SequenceFile
        val rdd = sc.sequenceFile[IntWritable, Text]("hdfs://path")
        读取Hadoop SequenceFile
        键值对RDD
    d.对象文件
        val rdd = sc.objectFile[MyClass]("hdfs://path")
        读取序列化对象
        Spark专用格式
    e.Hadoop InputFormat
        val rdd = sc.hadoopFile[K, V, InputFormat]("hdfs://path")
        val rdd = sc.newAPIHadoopFile[K, V, InputFormat]("hdfs://path")
        使用Hadoop InputFormat
        灵活读取

03.从其他RDD转换
    a.转换操作
        val rdd2 = rdd1.map(_ * 2)
        通过转换创建新RDD
        惰性执行
    b.依赖关系
        新RDD依赖父RDD
        记录血统信息
        用于容错

6.2 转换操作详解

01.map系列
    a.map
        rdd.map(x => x * 2)
        rdd.map(x => (x, x * x))
        一对一转换
        返回新RDD
    b.flatMap
        rdd.flatMap(x => 1 to x)
        rdd.flatMap(_.split(" "))
        一对多转换
        扁平化结果
    c.mapPartitions
        rdd.mapPartitions(iter => {
          val connection = createConnection()
          val result = iter.map(x => process(x, connection))
          connection.close()
          result
        })
        作用于整个分区
        减少函数调用
        可以初始化连接
    d.mapPartitionsWithIndex
        rdd.mapPartitionsWithIndex((index, iter) => {
          iter.map(x => (index, x))
        })
        带分区索引
        可以根据分区处理

02.filter和采样
    a.filter
        rdd.filter(x => x > 0)
        rdd.filter(_ % 2 == 0)
        筛选元素
        返回满足条件的RDD
    b.sample
        rdd.sample(withReplacement = false, fraction = 0.1)
        rdd.sample(withReplacement = true, fraction = 0.5, seed = 42)
        随机采样
        withReplacement:是否放回
        fraction:采样比例
    c.distinct
        rdd.distinct()
        rdd.distinct(numPartitions = 10)
        去重
        需要Shuffle

03.集合操作
    a.union
        rdd1.union(rdd2)
        rdd1 ++ rdd2
        合并RDD
        不去重
    b.intersection
        rdd1.intersection(rdd2)
        交集
        去重
        需要Shuffle
    c.subtract
        rdd1.subtract(rdd2)
        差集
        rdd1中不在rdd2中的元素
    d.cartesian
        rdd1.cartesian(rdd2)
        笛卡尔积
        rdd1和rdd2的所有组合

04.键值对转换
    a.mapValues
        pairRDD.mapValues(v => v * 2)
        只转换值
        保持键不变
    b.flatMapValues
        pairRDD.flatMapValues(v => v.split(","))
        值的扁平化映射
        保持键
    c.keys和values
        pairRDD.keys
        pairRDD.values
        提取键或值
    d.groupByKey
        pairRDD.groupByKey()
        按键分组
        返回(K, Iterable[V])
        需要Shuffle
        慎用,考虑reduceByKey
    e.reduceByKey
        pairRDD.reduceByKey(_ + _)
        pairRDD.reduceByKey((a, b) => a + b, numPartitions = 10)
        按键归约
        先本地聚合再Shuffle
        性能优于groupByKey
    f.aggregateByKey
        pairRDD.aggregateByKey(0)(
          (acc, v) => acc + v,
          (acc1, acc2) => acc1 + acc2
        )
        按键聚合
        可以改变类型
        两个函数:局部和全局
    g.combineByKey
        pairRDD.combineByKey(
          (v: Int) => (v, 1),
          (acc: (Int, Int), v: Int) => (acc._1 + v, acc._2 + 1),
          (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
        ).mapValues(acc => acc._1 / acc._2.toDouble)
        最灵活的聚合
        自定义累加器
        计算平均值等

05.Join操作
    a.join
        rdd1.join(rdd2)
        内连接
        返回(K, (V1, V2))
    b.leftOuterJoin
        rdd1.leftOuterJoin(rdd2)
        左外连接
        返回(K, (V1, Option[V2]))
    c.rightOuterJoin
        rdd1.rightOuterJoin(rdd2)
        右外连接
        返回(K, (Option[V1], V2))
    d.fullOuterJoin
        rdd1.fullOuterJoin(rdd2)
        全外连接
        返回(K, (Option[V1], Option[V2]))
    e.cogroup
        rdd1.cogroup(rdd2)
        协同分组
        返回(K, (Iterable[V1], Iterable[V2]))
        类似SQL的FULL OUTER JOIN

06.排序操作
    a.sortByKey
        pairRDD.sortByKey(ascending = true)
        pairRDD.sortByKey(ascending = false, numPartitions = 10)
        按键排序
        需要Shuffle
    b.sortBy
        rdd.sortBy(x => x, ascending = true)
        rdd.sortBy(_.length, ascending = false)
        自定义排序键
        灵活排序
    c.top和takeOrdered
        rdd.top(10)
        rdd.takeOrdered(10)(Ordering.by(-_))
        获取最大/最小的N个元素
        不需要全局排序

07.分区操作
    a.repartition
        rdd.repartition(10)
        改变分区数
        Shuffle操作
        可增可减
    b.coalesce
        rdd.coalesce(5, shuffle = false)
        减少分区数
        shuffle=false避免Shuffle
        优化性能
    c.partitionBy
        pairRDD.partitionBy(new HashPartitioner(10))
        pairRDD.partitionBy(new RangePartitioner(10, rdd))
        自定义分区器
        影响后续操作
    d.glom
        rdd.glom()
        将每个分区的元素组成数组
        返回Array[T]的RDD
        查看分区内容

6.3 [3]行动操作详解

01.聚合操作
    a.reduce
        rdd.reduce(_ + _)
        rdd.reduce((a, b) => math.max(a, b))
        聚合所有元素
        返回单个值
    b.fold
        rdd.fold(0)(_ + _)
        带初始值的归约
        每个分区使用初始值
    c.aggregate
        rdd.aggregate(0)(
          (acc, v) => acc + v,
          (acc1, acc2) => acc1 + acc2
        )
        两个函数:局部和全局
        可以改变类型
    d.treeReduce和treeAggregate
        rdd.treeReduce(_ + _)
        rdd.treeAggregate(0)(_ + _, _ + _)
        树形聚合
        减少Driver压力
        适合大规模数据

02.收集操作
    a.collect
        val array = rdd.collect()
        收集所有元素到Driver
        返回数组
        小心OOM
    b.take
        val array = rdd.take(10)
        获取前N个元素
        不需要计算所有分区
    c.first
        val elem = rdd.first()
        获取第一个元素
        等价于take(1).head
    d.top
        val array = rdd.top(10)
        获取最大的N个元素
        需要排序
    e.takeOrdered
        val array = rdd.takeOrdered(10)
        val array = rdd.takeOrdered(10)(Ordering.by(-_))
        获取最小的N个元素
        自定义排序
    f.takeSample
        val array = rdd.takeSample(withReplacement = false, num = 10)
        随机采样N个元素
        返回到Driver

03.统计操作
    a.count
        val n = rdd.count()
        元素个数
        触发计算
    b.countByKey
        val map = pairRDD.countByKey()
        每个键的元素个数
        返回Map
        结果收集到Driver
    c.countByValue
        val map = rdd.countByValue()
        每个值的出现次数
        返回Map
    d.countApprox
        val approx = rdd.countApprox(timeout = 1000)
        近似计数
        超时返回近似值
    e.sum,mean,stdev,variance
        rdd.sum()
        rdd.mean()
        rdd.stdev()
        rdd.variance()
        数值统计
        仅数值RDD
    f.stats
        val stats = rdd.stats()
        stats.count
        stats.mean
        stats.stdev
        stats.max
        stats.min
        统计摘要
        一次计算多个指标

04.保存操作
    a.saveAsTextFile
        rdd.saveAsTextFile("hdfs://path")
        rdd.saveAsTextFile("hdfs://path", classOf[GzipCodec])
        保存为文本文件
        支持压缩
    b.saveAsSequenceFile
        pairRDD.saveAsSequenceFile("hdfs://path")
        保存为SequenceFile
        键值对RDD
    c.saveAsObjectFile
        rdd.saveAsObjectFile("hdfs://path")
        保存为对象文件
        序列化存储
    d.saveAsHadoopFile
        pairRDD.saveAsHadoopFile("hdfs://path", classOf[K], classOf[V], classOf[OutputFormat])
        使用Hadoop OutputFormat
        灵活保存

05.遍历操作
    a.foreach
        rdd.foreach(println)
        rdd.foreach(x => writeToDatabase(x))
        对每个元素执行操作
        在Executor上执行
        无返回值
    b.foreachPartition
        rdd.foreachPartition(iter => {
          val connection = createConnection()
          iter.foreach(x => connection.write(x))
          connection.close()
        })
        对每个分区执行操作
        可以初始化连接
        批量处理

06.查找操作
    a.lookup
        pairRDD.lookup(key)
        查找键对应的所有值
        返回Seq[V]
    b.min和max
        rdd.min()
        rdd.max()
        最小值和最大值
        需要隐式Ordering

6.4 [4]持久化和Checkpoint

01.缓存级别
    a.MEMORY_ONLY
        rdd.cache()
        rdd.persist(StorageLevel.MEMORY_ONLY)
        仅内存存储
        反序列化对象
        最快但占内存
    b.MEMORY_AND_DISK
        rdd.persist(StorageLevel.MEMORY_AND_DISK)
        内存+磁盘
        内存不足溢写磁盘
        平衡性能和可靠性
    c.MEMORY_ONLY_SER
        rdd.persist(StorageLevel.MEMORY_ONLY_SER)
        序列化存储
        节省内存
        需要反序列化开销
    d.DISK_ONLY
        rdd.persist(StorageLevel.DISK_ONLY)
        仅磁盘存储
        节省内存
        性能较低
    e.OFF_HEAP
        rdd.persist(StorageLevel.OFF_HEAP)
        堆外内存
        避免GC
        需要配置
    f.带副本
        rdd.persist(StorageLevel.MEMORY_ONLY_2)
        rdd.persist(StorageLevel.MEMORY_AND_DISK_2)
        存储2个副本
        提高可靠性

02.缓存管理
    a.缓存
        rdd.cache()
        rdd.persist()
        缓存RDD
        加速后续操作
    b.释放缓存
        rdd.unpersist()
        rdd.unpersist(blocking = true)
        释放缓存
        及时清理
    c.查看缓存
        Spark UI Storage标签
        查看缓存RDD
        内存使用情况

03.Checkpoint
    a.设置checkpoint目录
        sc.setCheckpointDir("hdfs://path/to/checkpoint")
        指定checkpoint目录
        可靠存储
    b.checkpoint RDD
        rdd.checkpoint()
        持久化到可靠存储
        截断血统
    c.与缓存区别
        checkpoint写可靠存储
        cache写内存或本地磁盘
        checkpoint截断依赖
        cache保留依赖
    d.使用场景
        血统链太长
        宽依赖太多
        容错开销大
        迭代算法

04.持久化最佳实践
    a.何时缓存
        多次使用的RDD
        计算代价高的RDD
        迭代算法
        交互式查询
    b.选择存储级别
        频繁使用用MEMORY
        大数据用MEMORY_AND_DISK
        序列化节省空间
        根据场景选择
    c.及时释放
        不再使用时unpersist
        避免内存浪费
        监控内存使用
    d.checkpoint时机
        血统链长时checkpoint
        关键中间结果
        迭代算法中间状态

7 DataFrame和Dataset API详解

7.1 DataFrame创建和操作

01.创建DataFrame
    a.从RDD创建
        val rdd = sc.parallelize(Seq((1, "Alice", 25), (2, "Bob", 30)))
        val df = rdd.toDF("id", "name", "age")
        指定列名
        自动推断类型
    b.从集合创建
        val data = Seq((1, "Alice", 25), (2, "Bob", 30))
        val df = data.toDF("id", "name", "age")
        直接从集合
    c.从文件创建
        val df = spark.read.json("path/to/file.json")
        val df = spark.read.parquet("path/to/file.parquet")
        val df = spark.read.csv("path/to/file.csv")
        val df = spark.read.orc("path/to/file.orc")
        多种格式支持
    d.从数据源创建
        val df = spark.read
          .format("jdbc")
          .option("url", "jdbc:mysql://localhost:3306/db")
          .option("dbtable", "table")
          .option("user", "root")
          .option("password", "password")
          .load()
        JDBC数据源
        配置连接信息
    e.使用Schema
        import org.apache.spark.sql.types._
        val schema = StructType(Seq(
          StructField("id", IntegerType, nullable = false),
          StructField("name", StringType, nullable = true),
          StructField("age", IntegerType, nullable = true)
        ))
        val df = spark.read.schema(schema).json("path")
        显式指定Schema
        类型安全

02.选择和投影
    a.select
        df.select("name", "age")
        df.select($"name", $"age")
        df.select(col("name"), col("age"))
        df.select(expr("name"), expr("age + 1"))
        选择列
        多种语法
    b.selectExpr
        df.selectExpr("name", "age * 2 as doubled_age")
        df.selectExpr("avg(age)", "count(distinct name)")
        SQL表达式
        聚合函数
    c.drop
        df.drop("age")
        df.drop("age", "address")
        删除列
    d.withColumn
        df.withColumn("age_plus_one", $"age" + 1)
        df.withColumn("full_name", concat($"first_name", lit(" "), $"last_name"))
        添加或替换列
        表达式计算
    e.withColumnRenamed
        df.withColumnRenamed("name", "full_name")
        重命名列

03.过滤和条件
    a.filter/where
        df.filter($"age" > 18)
        df.filter("age > 18")
        df.where($"age" > 18 && $"name".startsWith("A"))
        条件过滤
        SQL表达式或Column表达式
    b.多条件
        df.filter($"age" > 18 && $"age" < 60)
        df.filter($"age".between(18, 60))
        df.filter($"name".isin("Alice", "Bob"))
        组合条件
    c.null处理
        df.filter($"age".isNotNull)
        df.filter($"age".isNull)
        df.na.drop()  // 删除包含null的行
        df.na.fill(0)  // 填充null
        df.na.fill(Map("age" -> 0, "name" -> "Unknown"))
        null值处理

04.聚合操作
    a.基本聚合
        df.count()
        df.select(count("*"))
        df.select(sum("age"), avg("age"), max("age"), min("age"))
        聚合函数
    b.groupBy
        df.groupBy("department").count()
        df.groupBy("department").agg(avg("salary"), max("age"))
        df.groupBy("department", "gender").count()
        分组聚合
    c.agg
        df.agg(
          count("*").as("total"),
          avg("age").as("avg_age"),
          max("salary").as("max_salary")
        )
        多个聚合
        命名结果
    d.pivot
        df.groupBy("year").pivot("month").sum("sales")
        数据透视
        行转列

05.排序操作
    a.orderBy/sort
        df.orderBy("age")
        df.orderBy($"age".desc)
        df.orderBy($"age".asc, $"name".desc)
        排序
        升序降序
    b.sortWithinPartitions
        df.sortWithinPartitions("age")
        分区内排序
        不全局排序
        性能更好

06.去重和采样
    a.distinct
        df.distinct()
        去重
        所有列
    b.dropDuplicates
        df.dropDuplicates()
        df.dropDuplicates("name")
        df.dropDuplicates("name", "age")
        指定列去重
    c.sample
        df.sample(withReplacement = false, fraction = 0.1)
        df.sample(withReplacement = true, fraction = 0.5, seed = 42)
        随机采样

7.2 Dataset操作

01.创建Dataset
    a.从序列创建
        case class Person(name: String, age: Int)
        val ds = Seq(Person("Alice", 25), Person("Bob", 30)).toDS()
        类型安全
        编译时检查
    b.从DataFrame转换
        val ds = df.as[Person]
        指定类型
        需要case class
    c.从RDD转换
        val ds = rdd.toDS()
        自动推断类型

02.类型安全操作
    a.map
        ds.map(p => p.name.toUpperCase)
        ds.map(p => (p.name, p.age * 2))
        类型安全转换
        编译时检查
    b.flatMap
        ds.flatMap(p => p.name.split(" "))
        扁平化映射
    c.filter
        ds.filter(p => p.age > 18)
        类型安全过滤
    d.mapPartitions
        ds.mapPartitions(iter => {
          val connection = createConnection()
          val result = iter.map(p => process(p, connection))
          connection.close()
          result
        })
        分区级别操作
        复用连接

03.SQL风格操作
    a.select
        ds.select($"name", $"age")
        ds.select(col("name"), col("age"))
        DataFrame风格
    b.where
        ds.where($"age" > 18)
        条件过滤
    c.groupBy
        ds.groupBy("department").count()
        ds.groupByKey(_.department).count()
        分组操作
        类型安全groupByKey

04.聚合操作
    a.groupByKey
        ds.groupByKey(_.department)
          .mapGroups((key, iter) => (key, iter.size))
        类型安全分组
        自定义聚合
    b.reduceGroups
        ds.groupByKey(_.department)
          .reduceGroups((p1, p2) =>
            Person(p1.name, p1.age + p2.age))
        分组归约
    c.agg
        ds.groupByKey(_.department)
          .agg(avg($"age").as[Double])
        分组聚合
        类型安全

05.Join操作
    a.join
        ds1.join(ds2, "id")
        ds1.join(ds2, ds1("id") === ds2("id"))
        内连接
    b.joinWith
        ds1.joinWith(ds2, ds1("id") === ds2("id"))
        返回(T1, T2)
        保持类型
    c.外连接
        ds1.join(ds2, Seq("id"), "left")
        ds1.join(ds2, Seq("id"), "right")
        ds1.join(ds2, Seq("id"), "outer")
        左右全外连接

06.转换
    a.toDF
        ds.toDF()
        ds.toDF("col1", "col2")
        转为DataFrame
        指定列名
    b.rdd
        ds.rdd
        转为RDD
        类型安全RDD

7.3 [3]SQL操作

01.注册表
    a.临时视图
        df.createOrReplaceTempView("people")
        注册临时表
        会话级别
    b.全局临时视图
        df.createGlobalTempView("people")
        全局可见
        跨会话
        使用global_temp.people访问
    c.注册函数
        spark.udf.register("addOne", (x: Int) => x + 1)
        注册UDF
        SQL中使用

02.执行SQL
    a.基本查询
        val result = spark.sql("SELECT * FROM people WHERE age > 18")
        标准SQL语法
        返回DataFrame
    b.复杂查询
        spark.sql("""
          SELECT department, AVG(salary) as avg_salary
          FROM employees
          GROUP BY department
          HAVING AVG(salary) > 50000
          ORDER BY avg_salary DESC
        """)
        聚合、过滤、排序
    c.Join查询
        spark.sql("""
          SELECT e.name, d.dept_name
          FROM employees e
          JOIN departments d ON e.dept_id = d.id
        """)
        多表Join
    d.子查询
        spark.sql("""
          SELECT * FROM employees
          WHERE salary > (SELECT AVG(salary) FROM employees)
        """)
        嵌套查询

03.DDL操作
    a.创建表
        spark.sql("""
          CREATE TABLE people (
            id INT,
            name STRING,
            age INT
          ) USING parquet
          PARTITIONED BY (age)
        """)
        DDL创建表
        指定格式和分区
    b.插入数据
        spark.sql("INSERT INTO people VALUES (1, 'Alice', 25)")
        spark.sql("INSERT INTO people SELECT * FROM source_table")
        插入数据
    c.删除表
        spark.sql("DROP TABLE IF EXISTS people")
        删除表

04.Catalog操作
    a.列出数据库
        spark.catalog.listDatabases().show()
        查看所有数据库
    b.列出表
        spark.catalog.listTables().show()
        spark.catalog.listTables("database_name").show()
        查看表
    c.列出列
        spark.catalog.listColumns("table_name").show()
        查看表结构
    d.缓存表
        spark.catalog.cacheTable("table_name")
        spark.catalog.uncacheTable("table_name")
        缓存和释放

7.4 [4]窗口函数

01.窗口定义
    a.基本窗口
        import org.apache.spark.sql.expressions.Window
        val windowSpec = Window.partitionBy("department").orderBy("salary")
        定义窗口
        分区和排序
    b.行范围窗口
        val windowSpec = Window
          .partitionBy("department")
          .orderBy("salary")
          .rowsBetween(Window.unboundedPreceding, Window.currentRow)
        基于行数
        累计窗口
    c.值范围窗口
        val windowSpec = Window
          .partitionBy("department")
          .orderBy("date")
          .rangeBetween(-86400, 0)
        基于值范围
        时间窗口

02.排名函数
    a.row_number
        df.withColumn("row_num", row_number().over(windowSpec))
        连续排名
        1,2,3,4
    b.rank
        df.withColumn("rank", rank().over(windowSpec))
        并列排名
        1,2,2,4
    c.dense_rank
        df.withColumn("dense_rank", dense_rank().over(windowSpec))
        密集排名
        1,2,2,3
    d.ntile
        df.withColumn("quartile", ntile(4).over(windowSpec))
        分桶
        将数据分成N组

03.聚合函数
    a.累计和
        df.withColumn("cumsum", sum("amount").over(windowSpec))
        累计求和
    b.移动平均
        val windowSpec = Window.orderBy("date").rowsBetween(-2, 0)
        df.withColumn("ma3", avg("value").over(windowSpec))
        3日移动平均
    c.累计统计
        df.withColumn("cumcount", count("*").over(windowSpec))
        df.withColumn("cummax", max("value").over(windowSpec))
        df.withColumn("cummin", min("value").over(windowSpec))
        累计计数、最大、最小

04.偏移函数
    a.lag
        df.withColumn("prev_value", lag("value", 1).over(windowSpec))
        df.withColumn("prev2_value", lag("value", 2, 0).over(windowSpec))
        前N行的值
        可以指定默认值
    b.lead
        df.withColumn("next_value", lead("value", 1).over(windowSpec))
        后N行的值
    c.first和last
        df.withColumn("first_value", first("value").over(windowSpec))
        df.withColumn("last_value", last("value").over(windowSpec))
        窗口内第一个和最后一个值

05.窗口应用
    a.TopN
        val windowSpec = Window.partitionBy("category").orderBy($"sales".desc)
        df.withColumn("rank", row_number().over(windowSpec))
          .filter($"rank" <= 10)
        每个分类的Top10
    b.同比环比
        val windowSpec = Window.partitionBy("product").orderBy("month")
        df.withColumn("prev_month", lag("sales", 1).over(windowSpec))
          .withColumn("growth", ($"sales" - $"prev_month") / $"prev_month")
        计算增长率
    c.累计占比
        val windowSpec = Window.partitionBy("department").orderBy("salary")
        df.withColumn("cumsum", sum("salary").over(windowSpec))
          .withColumn("total", sum("salary").over(Window.partitionBy("department")))
          .withColumn("pct", $"cumsum" / $"total")
        累计占比

7.5 [5]高级操作

01.Union和Join
    a.union
        df1.union(df2)
        df1.unionAll(df2)
        df1.unionByName(df2)
        合并DataFrame
        列数和类型要匹配
    b.join
        df1.join(df2, "id")
        df1.join(df2, Seq("id", "name"))
        df1.join(df2, df1("id") === df2("id"))
        多种Join方式
    c.join类型
        df1.join(df2, Seq("id"), "inner")
        df1.join(df2, Seq("id"), "left")
        df1.join(df2, Seq("id"), "right")
        df1.join(df2, Seq("id"), "outer")
        df1.join(df2, Seq("id"), "left_semi")
        df1.join(df2, Seq("id"), "left_anti")
        内连接、外连接、半连接、反连接

02.Explode和Pivot
    a.explode
        df.select($"name", explode($"hobbies").as("hobby"))
        展开数组
        一行变多行
    b.explode_outer
        df.select($"name", explode_outer($"hobbies").as("hobby"))
        展开数组
        保留null
    c.posexplode
        df.select($"name", posexplode($"hobbies").as(Seq("pos", "hobby")))
        展开数组
        带位置索引
    d.pivot
        df.groupBy("year").pivot("month").sum("sales")
        数据透视
        行转列
    e.unpivot
        df.selectExpr("year", "stack(12, 'Jan', jan, 'Feb', feb, ...) as (month, sales)")
        列转行
        反透视

03.UDF和UDAF
    a.注册UDF
        val addOne = udf((x: Int) => x + 1)
        df.select(addOne($"age"))
        定义UDF
        DataFrame中使用
    b.SQL UDF
        spark.udf.register("addOne", (x: Int) => x + 1)
        spark.sql("SELECT addOne(age) FROM people")
        注册到SQL
    c.UDAF
        import org.apache.spark.sql.expressions.Aggregator
        import org.apache.spark.sql.Encoder
        import org.apache.spark.sql.Encoders
        case class Average(var sum: Long, var count: Long)
        object MyAverage extends Aggregator[Long, Average, Double] {
          def zero: Average = Average(0L, 0L)
          def reduce(buffer: Average, data: Long): Average = {
            buffer.sum += data
            buffer.count += 1
            buffer
          }
          def merge(b1: Average, b2: Average): Average = {
            b1.sum += b2.sum
            b1.count += b2.count
            b1
          }
          def finish(reduction: Average): Double = {
            reduction.sum.toDouble / reduction.count
          }
          def bufferEncoder: Encoder[Average] = Encoders.product
          def outputEncoder: Encoder[Double] = Encoders.scalaDouble
        }
        val ds = df.as[Long]
        val avgCol = MyAverage.toColumn.name("average")
        ds.select(avgCol)
        自定义聚合函数
        类型安全

04.缓存和持久化
    a.cache
        df.cache()
        df.persist()
        缓存DataFrame
        加速查询
    b.存储级别
        df.persist(StorageLevel.MEMORY_AND_DISK)
        df.persist(StorageLevel.MEMORY_ONLY)
        选择存储级别
    c.释放缓存
        df.unpersist()
        释放内存
    d.查看缓存
        Spark UI Storage标签
        监控缓存使用

05.分区和重分区
    a.repartition
        df.repartition(10)
        df.repartition($"department")
        df.repartition(10, $"department")
        改变分区数
        按列分区
    b.coalesce
        df.coalesce(5)
        减少分区数
        避免Shuffle
    c.partitionBy写入
        df.write.partitionBy("year", "month").parquet("path")
        分区写入
        Hive风格分区

8 高级特性

8.1 广播变量

01.广播变量概念
    a.定义
        只读共享变量
        广播到所有节点
        避免重复传输
        节省网络和内存
    b.工作原理
        Driver创建广播变量
        发送到所有Executor
        Executor本地缓存
        任务共享使用
    c.适用场景
        大型只读数据
        查找表
        配置信息
        小表Join

02.使用广播变量
    a.创建广播
        val broadcastVar = sc.broadcast(Array(1, 2, 3))
        广播数组
        返回Broadcast对象
    b.访问广播
        val value = broadcastVar.value
        获取广播的值
        在Executor上访问
    c.销毁广播
        broadcastVar.unpersist()  // 异步删除
        broadcastVar.destroy()  // 同步删除
        释放资源
        及时清理

03.广播Join
    a.小表广播
        val smallTable = sc.broadcast(smallData.collectAsMap())
        largeRDD.map { case (k, v) =>
          (k, v, smallTable.value.get(k))
        }
        避免Shuffle
        提高性能
    b.自动广播
        spark.sql.autoBroadcastJoinThreshold
        默认10MB
        自动广播小表
        SQL优化
    c.手动广播
        import org.apache.spark.sql.functions.broadcast
        df1.join(broadcast(df2), "key")
        强制广播
        控制Join策略

04.广播最佳实践
    a.大小控制
        广播变量不宜过大
        一般小于几百MB
        过大影响性能
        考虑网络和内存
    b.复用广播
        多个操作共享
        避免重复广播
        及时清理
    c.监控
        Spark UI查看
        广播大小
        传输时间

8.2 累加器

01.累加器概念
    a.定义
        共享累加变量
        只能累加
        Driver读取结果
        Executor只能写
    b.工作原理
        Executor累加值
        发送到Driver
        Driver聚合结果
        保证一致性
    c.适用场景
        计数器
        求和
        自定义聚合
        调试统计

02.使用累加器
    a.创建累加器
        val accum = sc.longAccumulator("My Accumulator")
        创建Long累加器
        命名累加器
    b.累加操作
        rdd.foreach(x => accum.add(x))
        在Executor中累加
        只能add操作
    c.读取结果
        val result = accum.value
        在Driver中读取
        获取累加结果

03.自定义累加器
    a.继承AccumulatorV2
        class VectorAccumulator extends AccumulatorV2[Vector, Vector] {
          private var sum = Vector.zeros(3)
          def isZero: Boolean = sum == Vector.zeros(3)
          def copy(): VectorAccumulator = new VectorAccumulator
          def reset(): Unit = { sum = Vector.zeros(3) }
          def add(v: Vector): Unit = { sum += v }
          def merge(other: VectorAccumulator): Unit = { sum += other.sum }
          def value: Vector = sum
        }
        实现抽象方法
        自定义累加逻辑
    b.注册累加器
        val vecAccum = new VectorAccumulator
        sc.register(vecAccum, "VectorAccumulator")
        注册到SparkContext
        可以在UI中查看

04.累加器注意事项
    a.转换操作中使用
        转换操作惰性执行
        可能多次执行
        累加器可能重复计数
        推荐在行动操作中使用
    b.容错
        任务失败重试
        累加器可能重复累加
        exactly-once不保证
        仅用于调试统计
    c.性能
        累加器有网络开销
        不要过度使用
        批量累加

8.3 Shuffle机制

01.Shuffle概念
    a.定义
        数据重新分区
        跨节点数据传输
        磁盘IO和网络IO
        性能瓶颈
    b.触发Shuffle的操作
        reduceByKey,groupByKey
        join,cogroup
        distinct,repartition
        sortByKey
    c.Shuffle过程
        Map端:分区,排序,写磁盘
        Reduce端:拉取,合并,计算
        中间文件管理

02.Shuffle Write
    a.分区
        根据分区器计算分区
        HashPartitioner或RangePartitioner
        数据分组
    b.排序
        按分区ID排序
        可选按Key排序
        Sort Shuffle
    c.写磁盘
        缓冲区满溢写
        生成Shuffle文件
        索引文件

03.Shuffle Read
    a.拉取数据
        从Map端拉取数据
        网络传输
        失败重试
    b.合并数据
        合并多个Map输出
        排序合并
        聚合计算
    c.内存管理
        内存缓冲区
        溢写磁盘
        避免OOM

04.Shuffle优化
    a.减少Shuffle
        使用reduceByKey代替groupByKey
        使用mapPartitions批处理
        广播小表
        预分区
    b.调整参数
        spark.shuffle.file.buffer:写缓冲,默认32KB
        spark.reducer.maxSizeInFlight:读缓冲,默认48MB
        spark.shuffle.sort.bypassMergeThreshold:bypass阈值,默认200
        spark.shuffle.compress:压缩,默认true
    c.选择Shuffle方式
        Sort Shuffle:默认,适合大多数场景
        Tungsten Sort:优化的Sort
        Bypass Sort:分区少时bypass排序
    d.监控Shuffle
        Spark UI查看
        Shuffle读写量
        Shuffle时间
        定位瓶颈

8.4 任务调度

01.调度层次
    a.Application
        用户提交的应用
        包含多个Job
        SparkContext管理
    b.Job
        行动操作触发
        包含多个Stage
        DAGScheduler调度
    c.Stage
        根据Shuffle划分
        包含多个Task
        并行执行
    d.Task
        最小执行单元
        作用于一个分区
        Executor执行

02.DAGScheduler
    a.DAG构建
        根据RDD依赖构建DAG
        有向无环图
        表示计算流程
    b.Stage划分
        宽依赖划分Stage边界
        窄依赖在同一Stage
        Stage之间有依赖关系
        拓扑排序
    c.提交Stage
        按依赖顺序提交
        父Stage先执行
        子Stage等待父Stage完成
        TaskScheduler调度Task

03.TaskScheduler
    a.Task分配
        将Task分配到Executor
        考虑数据本地性
        负载均衡
        资源可用性
    b.数据本地性
        PROCESS_LOCAL:同JVM
        NODE_LOCAL:同节点
        RACK_LOCAL:同机架
        ANY:任意位置
        优先本地性高的
    c.推测执行
        检测慢Task
        启动备份Task
        先完成的有效
        spark.speculation.enabled

04.调度策略
    a.FIFO调度
        先进先出
        简单公平
        默认策略
        适合单用户
    b.FAIR调度
        公平调度
        多用户共享
        调度池
        权重配置
    c.调度池
        创建调度池
        设置权重和最小资源
        任务分配到池
        池间公平调度

05.动态资源分配
    a.概念
        根据负载动态调整Executor数量
        空闲时释放
        繁忙时申请
        提高资源利用率
    b.配置
        spark.dynamicAllocation.enabled:true
        spark.dynamicAllocation.minExecutors:最小Executor
        spark.dynamicAllocation.maxExecutors:最大Executor
        spark.dynamicAllocation.executorIdleTimeout:空闲超时
    c.适用场景
        共享集群
        负载波动大
        多用户环境
        提高资源利用率

8.5 内存调优

01.内存配置
    a.Executor内存
        spark.executor.memory:Executor堆内存
        spark.executor.memoryOverhead:堆外内存
        总内存=堆内存+堆外内存
        根据数据量配置
    b.Driver内存
        spark.driver.memory:Driver堆内存
        spark.driver.memoryOverhead:堆外内存
        collect操作需要足够内存
    c.内存分配
        Storage Memory:缓存RDD
        Execution Memory:任务执行
        User Memory:用户数据结构
        Reserved Memory:系统保留

02.Storage内存优化
    a.选择存储级别
        MEMORY_ONLY:最快,占内存
        MEMORY_AND_DISK:平衡
        MEMORY_ONLY_SER:节省内存
        根据场景选择
    b.缓存管理
        cache()频繁使用的RDD
        unpersist()及时释放
        监控缓存使用
        避免缓存过多
    c.序列化
        使用Kryo序列化
        spark.serializer:org.apache.spark.serializer.KryoSerializer
        注册类
        减少内存占用

03.Execution内存优化
    a.Shuffle内存
        spark.shuffle.memoryFraction:Shuffle内存比例
        增大减少溢写
        平衡内存和磁盘
    b.聚合内存
        spark.sql.shuffle.partitions:Shuffle分区数,默认200
        增加分区减少单分区内存
        避免OOM
    c.Join内存
        使用Broadcast Join
        小表广播
        避免Shuffle
        节省内存

04.GC调优
    a.选择GC
        使用G1 GC
        -XX:+UseG1GC
        适合大堆内存
        减少Full GC
    b.GC参数
        -XX:MaxGCPauseMillis:最大暂停时间
        -XX:InitiatingHeapOccupancyPercent:GC触发阈值
        -XX:G1HeapRegionSize:Region大小
        根据应用调整
    c.监控GC
        查看GC日志
        -XX:+PrintGCDetails
        -XX:+PrintGCTimeStamps
        分析GC行为
        优化参数

05.内存溢出处理
    a.增加内存
        增大executor内存
        增加executor数量
        扩展集群资源
    b.减少内存使用
        减少缓存
        增加分区数
        使用序列化
        过滤不需要的数据
    c.优化算法
        避免数据倾斜
        使用高效算子
        减少Shuffle
        优化Join策略

8.6 数据倾斜处理

01.数据倾斜识别
    a.现象
        某些Task执行时间特别长
        Shuffle读写不均衡
        部分Executor内存溢出
        整体性能差
    b.定位
        Spark UI查看Task时间
        查看Shuffle读写量
        检查数据分布
        分析Key分布
    c.原因
        Key分布不均
        数据源倾斜
        Join键倾斜
        聚合键倾斜

02.数据倾斜解决方案
    a.过滤异常数据
        过滤null或异常Key
        单独处理
        减少倾斜
    b.增加并行度
        增加Shuffle分区数
        spark.sql.shuffle.partitions
        分散数据
        减轻单Task压力
    c.自定义分区器
        class CustomPartitioner extends Partitioner {
          def getPartition(key: Any) = {
            // 自定义分区逻辑,打散热点Key
          }
        }
        打散热点Key
        均匀分布
    d.两阶段聚合
        第一阶段:加随机前缀,局部聚合
        第二阶段:去前缀,全局聚合
        打散数据
        减少倾斜

03.Join倾斜优化
    a.Broadcast Join
        小表广播
        避免Shuffle
        适合小表Join大表
    b.Salting
        大表加盐(随机前缀)
        小表扩容(复制多份)
        打散Join
        均匀分布
    c.拆分Join
        倾斜Key单独Join
        非倾斜Key正常Join
        Union结果
        分而治之
    d.Map端Join
        使用mapPartitions
        分区内Join
        避免Shuffle
        适合特定场景

04.倾斜预防
    a.数据预处理
        清洗异常数据
        过滤无效Key
        数据采样分析
        提前发现问题
    b.合理设计Key
        避免热点Key
        组合Key
        哈希Key
        均匀分布
    c.监控告警
        监控Task时间
        监控数据分布
        及时发现倾斜
        快速响应

9 使用实战

9.1 环境搭建

01.本地模式
    a.下载安装
        从spark.apache.org下载
        解压到指定目录
        配置SPARK_HOME环境变量
        添加bin到PATH
    b.验证安装
        spark-shell  // Scala Shell
        pyspark  // Python Shell
        spark-submit --version
        查看版本信息
    c.运行示例
        spark-submit --class org.apache.spark.examples.SparkPi \\
          --master local \\
          $SPARK_HOME/examples/jars/spark-examples_2.12-3.4.0.jar 10
        本地运行示例

02.Standalone集群
    a.配置Master
        cp conf/spark-env.sh.template conf/spark-env.sh
        编辑spark-env.sh:
        SPARK_MASTER_HOST=master-host
        SPARK_MASTER_PORT=7077
        启动Master:sbin/start-master.sh
    b.配置Worker
        编辑spark-env.sh:
        SPARK_MASTER_HOST=master-host
        启动Worker:sbin/start-worker.sh spark://master-host:7077
    c.提交应用
        spark-submit --master spark://master-host:7077 \\
          --class MainClass \\
          app.jar

03.YARN集群
    a.配置Hadoop
        确保Hadoop集群正常运行
        配置HADOOP_CONF_DIR
        Spark自动读取Hadoop配置
    b.Client模式
        spark-submit --master yarn \\
          --deploy-mode client \\
          --class MainClass \\
          app.jar
        Driver在客户端
        适合交互式
    c.Cluster模式
        spark-submit --master yarn \\
          --deploy-mode cluster \\
          --class MainClass \\
          app.jar
        Driver在YARN容器
        适合生产环境

04.开发环境
    a.IDE配置
        IntelliJ IDEA安装Scala插件
        创建Maven或SBT项目
        添加Spark依赖
        配置运行配置
    b.依赖配置
        Maven:
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.12</artifactId>
          <version>3.4.0</version>
        </dependency>
        SBT:
        libraryDependencies += "org.apache.spark" %% "spark-core" % "3.4.0"
    c.本地调试
        设置master为local[*]
        在IDE中运行
        断点调试
        查看日志

9.2 集群部署

01.资源规划
    a.硬件配置
        CPU:多核,推荐16核以上
        内存:大内存,推荐128GB以上
        磁盘:SSD优先,多块磁盘
        网络:万兆网络
    b.节点规划
        Master节点:1-2个,高可用
        Worker节点:根据数据量
        每个节点配置相同
        预留资源
    c.资源分配
        每个Executor:4-8核,16-32GB内存
        每个节点:2-4个Executor
        预留系统资源
        动态资源分配

02.高可用配置
    a.Master HA
        使用ZooKeeper
        配置spark.deploy.recoveryMode:ZOOKEEPER
        配置spark.deploy.zookeeper.url
        多个Master
    b.Metastore HA
        使用外部Metastore
        MySQL或PostgreSQL
        主备配置
        数据持久化
    c.HDFS HA
        Hadoop HA配置
        NameNode HA
        数据冗余
        自动故障转移

03.安全配置
    a.认证
        启用Kerberos
        spark.authenticate:true
        spark.authenticate.secret
        用户认证
    b.加密
        spark.network.crypto.enabled:true
        spark.io.encryption.enabled:true
        数据传输加密
        Shuffle加密
    c.授权
        Spark ACL
        HDFS权限
        Hive权限
        细粒度控制

04.监控部署
    a.Spark History Server
        配置spark.eventLog.enabled:true
        配置spark.eventLog.dir
        启动History Server
        查看历史应用
    b.Metrics配置
        配置metrics.properties
        输出到Ganglia,Graphite
        JMX监控
        自定义Metrics
    c.日志配置
        配置log4j.properties
        日志级别
        日志滚动
        集中日志收集

9.3 开发实战

01.WordCount示例
    a.Scala版本
        val spark = SparkSession.builder()
          .appName("WordCount")
          .master("local[*]")
          .getOrCreate()
        val lines = spark.read.textFile("input.txt")
        val words = lines.flatMap(_.split(" "))
        val wordCounts = words.groupBy("value").count()
        wordCounts.show()
        spark.stop()
        经典示例
    b.优化版本
        val lines = spark.sparkContext.textFile("input.txt")
        val counts = lines.flatMap(_.split(" "))
          .map((_, 1))
          .reduceByKey(_ + _)
        counts.saveAsTextFile("output")
        使用RDD API
        reduceByKey更高效

02.日志分析
    a.需求
        分析Web服务器日志
        统计访问量,UV,PV
        Top URL
        错误率
    b.实现
        val logs = spark.read.text("access.log")
        val parsed = logs.select(
          regexp_extract($"value", "^([^ ]*)", 1).as("ip"),
          regexp_extract($"value", "\\\"([^ ]*)\"", 1).as("url"),
          regexp_extract($"value", "\" (\\d{3})", 1).cast("int").as("status")
        )
        val pv = parsed.count()
        val uv = parsed.select("ip").distinct().count()
        val topUrls = parsed.groupBy("url").count()
          .orderBy($"count".desc).limit(10)
        val errorRate = parsed.filter($"status" >= 400).count() * 1.0 / pv
        日志解析和统计
    c.优化
        缓存parsed DataFrame
        分区存储
        增量处理
        定时任务

03.实时推荐
    a.需求
        实时计算用户行为
        更新推荐模型
        实时推荐商品
    b.实现
        val spark = SparkSession.builder()
          .appName("RealtimeRecommend")
          .getOrCreate()
        val events = spark.readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "localhost:9092")
          .option("subscribe", "user-events")
          .load()
        val parsed = events.selectExpr("CAST(value AS STRING)")
          .select(from_json($"value", schema).as("data"))
          .select("data.*")
        val recommendations = parsed
          .groupBy("userId")
          .agg(collect_list("itemId").as("items"))
          .select($"userId", explode(recommend_udf($"items")).as("recommendedItems"))
        val query = recommendations.writeStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "localhost:9092")
          .option("topic", "recommendations")
          .option("checkpointLocation", "/tmp/checkpoint")
          .start()
        query.awaitTermination()
        实时流处理

04.机器学习Pipeline
    a.需求
        训练分类模型
        特征工程
        模型评估
        模型持久化
    b.实现
        import org.apache.spark.ml.Pipeline
        import org.apache.spark.ml.classification.LogisticRegression
        import org.apache.spark.ml.feature.{VectorAssembler, StandardScaler}
        val data = spark.read.format("libsvm").load("data.txt")
        val Array(training, test) = data.randomSplit(Array(0.8, 0.2))
        val assembler = new VectorAssembler()
          .setInputCols(Array("feature1", "feature2"))
          .setOutputCol("features")
        val scaler = new StandardScaler()
          .setInputCol("features")
          .setOutputCol("scaledFeatures")
        val lr = new LogisticRegression()
          .setFeaturesCol("scaledFeatures")
          .setLabelCol("label")
        val pipeline = new Pipeline().setStages(Array(assembler, scaler, lr))
        val model = pipeline.fit(training)
        val predictions = model.transform(test)
        val evaluator = new BinaryClassificationEvaluator()
        val auc = evaluator.evaluate(predictions)
        model.write.overwrite().save("model")
        完整Pipeline

9.4 性能调优

01.应用级调优
    a.并行度
        spark.default.parallelism:默认并行度
        spark.sql.shuffle.partitions:SQL Shuffle分区
        设置为CPU核心数的2-3倍
        避免任务过多或过少
    b.序列化
        使用Kryo序列化
        spark.serializer:org.apache.spark.serializer.KryoSerializer
        注册类:spark.kryo.registrationRequired:true
        减少序列化开销
    c.压缩
        spark.rdd.compress:true
        spark.shuffle.compress:true
        spark.io.compression.codec:snappy
        减少存储和网络

02.内存调优
    a.Executor内存
        spark.executor.memory:根据数据量
        spark.executor.memoryOverhead:堆外内存
        spark.memory.fraction:统一内存比例
        spark.memory.storageFraction:Storage比例
    b.缓存策略
        cache()频繁使用的RDD
        选择合适的存储级别
        unpersist()及时释放
        监控缓存使用
    c.GC调优
        使用G1 GC
        spark.executor.extraJavaOptions:-XX:+UseG1GC
        调整GC参数
        监控GC日志

03.Shuffle调优
    a.减少Shuffle
        使用reduceByKey代替groupByKey
        使用Broadcast Join
        预分区
        合并小文件
    b.Shuffle参数
        spark.shuffle.file.buffer:32k-64k
        spark.reducer.maxSizeInFlight:48m-96m
        spark.shuffle.sort.bypassMergeThreshold:200
        根据场景调整
    c.Shuffle服务
        启用External Shuffle Service
        spark.shuffle.service.enabled:true
        Executor失败不影响Shuffle数据
        提高稳定性

04.SQL调优
    a.CBO优化
        ANALYZE TABLE收集统计信息
        spark.sql.cbo.enabled:true
        spark.sql.statistics.histogram.enabled:true
        基于成本优化
    b.缓存表
        CACHE TABLE tableName
        频繁查询的表
        加速查询
        及时清理
    c.分区裁剪
        使用分区表
        WHERE条件包含分区列
        减少数据扫描
        提高性能
    d.列裁剪
        只SELECT需要的列
        减少数据传输
        提高性能

05.数据倾斜调优
    a.识别倾斜
        Spark UI查看Task时间
        检查数据分布
        分析Key分布
    b.解决方案
        过滤异常Key
        增加并行度
        自定义分区器
        两阶段聚合
        Salting技术
    c.Join优化
        Broadcast Join小表
        Salting打散大表
        拆分倾斜Key
        Map端Join

9.5 监控运维

01.Spark UI监控
    a.Jobs页面
        查看Job列表
        Job执行时间
        Stage信息
        DAG可视化
    b.Stages页面
        Stage执行时间
        Task分布
        Shuffle读写量
        定位慢Task
    c.Storage页面
        RDD缓存信息
        缓存大小
        内存使用
        淘汰情况
    d.Executors页面
        Executor列表
        资源使用
        GC时间
        Shuffle统计

02.History Server
    a.配置
        spark.eventLog.enabled:true
        spark.eventLog.dir:hdfs://path
        spark.history.fs.logDirectory:hdfs://path
        持久化事件日志
    b.启动
        sbin/start-history-server.sh
        访问http://host:18080
        查看历史应用
    c.应用
        分析历史任务
        性能对比
        问题排查
        趋势分析

03.Metrics监控
    a.配置Metrics
        复制metrics.properties.template
        配置Sink:Ganglia,Graphite,JMX
        启用Metrics
    b.监控指标
        JVM Metrics:内存,GC
        Executor Metrics:任务,Shuffle
        Driver Metrics:DAG,Job
        自定义Metrics
    c.可视化
        Grafana展示
        实时监控
        告警配置
        趋势分析

04.日志管理
    a.日志配置
        配置log4j.properties
        设置日志级别
        日志滚动
        日志格式
    b.日志收集
        使用Flume收集
        使用Logstash收集
        集中存储
        ELK分析
    c.日志分析
        错误日志分析
        性能日志分析
        定位问题
        优化应用

05.告警配置
    a.告警指标
        任务失败
        执行时间过长
        内存溢出
        GC频繁
        数据倾斜
    b.告警方式
        邮件告警
        短信告警
        企业微信/钉钉
        PagerDuty
    c.告警处理
        及时响应
        分析原因
        快速恢复
        总结经验

9.6 常见问题

01.内存问题
    a.Executor OOM
        原因:数据倾斜,缓存过多,分区过少
        解决:增加内存,增加分区,减少缓存,处理倾斜
        预防:合理配置,监控内存,优化代码
    b.Driver OOM
        原因:collect过多数据,广播变量过大
        解决:增加Driver内存,避免collect,减小广播
        预防:不要collect大数据,限制广播大小
    c.GC频繁
        原因:内存不足,对象过多,Full GC
        解决:增加内存,使用G1 GC,调整GC参数
        预防:优化代码,减少对象创建,序列化

02.性能问题
    a.任务慢
        原因:数据倾斜,Shuffle多,分区不合理
        解决:处理倾斜,减少Shuffle,调整分区
        预防:合理设计,监控性能,及时优化
    b.Shuffle慢
        原因:数据量大,网络慢,磁盘慢
        解决:减少Shuffle,优化网络,使用SSD
        预防:避免不必要Shuffle,预分区,压缩
    c.读写慢
        原因:小文件多,格式不优,分区不合理
        解决:合并小文件,使用Parquet,合理分区
        预防:选择合适格式,定期合并,分区存储

03.稳定性问题
    a.任务失败
        原因:数据问题,资源不足,代码bug
        解决:检查数据,增加资源,修复bug
        预防:数据校验,充足资源,充分测试
    b.Executor丢失
        原因:OOM,节点故障,网络问题
        解决:增加内存,修复节点,检查网络
        预防:监控资源,高可用,网络优化
    c.数据丢失
        原因:checkpoint失败,HDFS问题
        解决:检查checkpoint,修复HDFS
        预防:可靠存储,定期备份,监控

04.配置问题
    a.资源配置
        Executor数量,内存,核心数
        根据数据量和集群资源
        动态资源分配
        监控资源使用
    b.参数调优
        并行度,Shuffle分区,内存比例
        根据应用特点
        逐步调整
        压力测试验证
    c.兼容性
        Spark版本,Hadoop版本,Scala版本
        依赖冲突
        测试兼容性
        升级注意事项

05.调试技巧
    a.本地调试
        使用local模式
        IDE断点调试
        查看日志
        小数据集测试
    b.日志分析
        查看Driver日志
        查看Executor日志
        分析错误堆栈
        定位问题代码
    c.UI分析
        Spark UI查看
        Stage和Task分析
        Timeline分析
        定位性能瓶颈
    d.采样调试
        sample()采样数据
        take()查看数据
        count()验证数据量
        逐步调试