📚 课程目录第3章 Spark环境搭建与配置

3. Spark环境搭建与配置：集群模式选择、Session初始化、核心参数调优

高频行情数据处理对延迟和吞吐量有极高要求。Spark环境的搭建与配置直接影响标签计算的稳定性与效率。本节从集群模式选择、Session初始化、核心参数调优三个层面展开，帮助你在生产环境中快速搭建适配高频行情场景的Spark运行环境。

3.1 集群模式选择

Spark支持多种集群部署模式，不同模式在资源管理、任务调度、网络开销上存在显著差异。高频行情场景下，推荐优先考虑以下两种模式：

模式	资源管理器	适用场景	延迟特征	推荐指数
YARN 模式	Hadoop YARN	已有Hadoop集群，行情数据存储在HDFS	中等（需与YARN RM交互）	★★★★☆
Kubernetes 模式	K8s	容器化部署，弹性伸缩，微服务架构	低（容器直接调度）	★★★★★
Standalone 模式	Spark内置	小规模测试或开发环境	低（无额外调度开销）	★★★☆☆
Local 模式	单机多线程	本地调试、小批量验证	极低（无网络开销）	★★☆☆☆

实战建议：生产环境高频行情标签计算，优先选择 Kubernetes 模式。K8s 支持动态资源分配、快速扩缩容，且容器启动时间远低于 YARN 的 ApplicationMaster 申请流程，适合分钟级甚至秒级的行情批处理窗口。

3.2 SparkSession 初始化

SparkSession 是 Spark 2.0+ 的统一入口点，整合了 SQL、Streaming、MLlib 等模块。在高频行情标签计算中，Session 的初始化需要显式指定序列化器、动态分配开关、以及行情特有的时间语义配置。

3.2.1 基础初始化模板

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HighFreqTickLabel") \
    .master("k8s://https://k8s-api.example.com:6443") \
    .config("spark.kubernetes.container.image", "registry.example.com/spark:3.5.0-hft") \
    .config("spark.kubernetes.namespace", "hft-spark") \
    .config("spark.sql.session.timeZone", "Asia/Shanghai") \
    .config("spark.sql.shuffle.partitions", "200") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", "com.example.hft.TickDataRegistrator") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "2") \
    .config("spark.dynamicAllocation.maxExecutors", "20") \
    .config("spark.dynamicAllocation.initialExecutors", "4") \
    .getOrCreate()

  关键配置说明：
  spark.sql.session.timeZone：必须与行情数据的时间戳时区一致，避免标签计算中的时间偏移错误。
spark.serializer：Kryo 序列化比 Java 序列化快 10 倍以上，高频行情场景必须启用。
spark.dynamicAllocation.enabled：动态分配 Executor，适应行情高峰低谷的资源需求变化。

3.2.2 高频行情专用配置项

配置项	推荐值	说明
`spark.sql.shuffle.partitions`	200 ~ 500	根据数据量调整，行情数据量大时建议 500+
`spark.sql.adaptive.enabled`	true	开启自适应查询执行，自动合并小分区
`spark.sql.adaptive.coalescePartitions.enabled`	true	自动合并 Shuffle 后的小分区，减少文件数
`spark.sql.adaptive.skewJoin.enabled`	true	自动处理数据倾斜，行情数据中热门股票易倾斜
`spark.sql.adaptive.skewJoin.skewedPartitionFactor`	5	倾斜因子，超过中位数 5 倍即视为倾斜分区

3.3 核心参数调优

高频行情标签计算涉及大量 Shuffle 操作（如按股票代码分组计算指标、时间窗口聚合），Shuffle 分区和内存配置是性能瓶颈的关键。

3.3.1 Shuffle 分区调优

Shuffle 分区数直接影响任务并行度和文件输出数量。分区数过少会导致单个任务处理数据量过大，引发 OOM；分区数过多则增加调度开销和文件写入压力。

经验公式：分区数 = 集群总核心数 × 2 ~ 3
数据量参考：每个分区处理 100MB ~ 200MB 数据为宜
行情特殊场景：按股票代码分组时，热门股票（如贵州茅台、腾讯控股）数据量可能是冷门股票的 100 倍以上，必须开启 spark.sql.adaptive.skewJoin.enabled

# 动态调整 Shuffle 分区数（在代码中按需设置）
spark.conf.set("spark.sql.shuffle.partitions", 400)

# 查看当前分区数
print(spark.conf.get("spark.sql.shuffle.partitions"))

3.3.2 内存配置

Spark 内存分为 Execution（执行）、Storage（存储）和 Reserved（预留）三部分。高频行情场景下，Execution 内存占比应适当提高，因为标签计算以 CPU 密集型聚合为主。

参数	推荐值	说明
`spark.executor.memory`	8g ~ 32g	每个 Executor 的 JVM 堆内存
`spark.executor.memoryOverhead`	2g ~ 4g	堆外内存，用于 JVM 开销、网络缓冲区等
`spark.memory.fraction`	0.7	堆内存中用于 Execution + Storage 的比例
`spark.memory.storageFraction`	0.3	Storage 占 (Execution+Storage) 的比例，剩余 0.7 给 Execution
`spark.executor.cores`	2 ~ 4	每个 Executor 的 CPU 核心数，建议不超过 4

常见陷阱：不要将 spark.executor.memory 设置过大（如 64g+），会导致 GC 停顿时间过长，反而降低吞吐量。高频行情场景推荐每个 Executor 分配 8g ~ 16g 内存，配合 2 ~ 4 个核心，通过增加 Executor 数量来扩展总资源。

3.4 参数调优检查清单

在启动高频行情标签作业前，建议逐项检查以下配置：

序列化：已启用 Kryo 序列化并注册自定义类
动态分配：已开启，并设置合理的 min/max Executor 数量
Shuffle 分区：已根据数据量和集群规模设置
自适应执行：已开启 AQE，并启用倾斜 Join 处理
内存比例：Execution 占比不低于 0.7
时区配置：与行情数据时区一致
堆外内存：已设置 spark.executor.memoryOverhead，避免 OOM

快速验证：使用 spark.sparkContext.getConf().getAll 打印所有配置项，检查关键参数是否生效。首次运行建议使用小批量数据（如 1 分钟行情）测试配置正确性，再逐步扩展到全量数据。