Spark 批处理

盘后要重算全市场历史 K 线，或者给过去多年的数据增加一个新因子时，怎样把大任务拆开并行执行，而且某台机器失败后不用全部从头开始？

Driver、Executor 与集群管理器

Driver 是任务的协调者：读取程序、生成执行计划、把任务分发出去并汇总结果。Executor 在工作节点上真正读取数据和计算。Cluster Manager（例如 YARN、Kubernetes 或 Standalone）负责分配资源。

分区与惰性执行

Spark 会把数据拆成多个 partition，每个任务处理其中一部分。存储分区决定需要读取哪些文件，计算分区决定同时运行多少个任务，两者有关但不是同一个概念。

filter、select、groupBy 等 transformation 先记录计算计划；count、show、write 等 action 才触发执行。这样 Spark 可以统一优化，例如只读需要的列和日期分区。

Shuffle、依赖与 Stage

按股票代码 groupBy("code") 时，同一代码的数据要被重新分到同一个任务，这种跨分区重分发叫 Shuffle，通常涉及网络和磁盘，是批处理的高成本步骤。过滤、增加列等不需要跨分区交换的操作属于窄依赖；需要 Shuffle 的属于宽依赖。Spark 通常以 Shuffle 为边界切分 Stage。

Spark 还会记录数据从输入到结果经历的计算关系（lineage）。某个 Executor 失败时，系统可以重算丢失的分区，而不必重跑整个作业。自适应查询执行（AQE）则会根据运行中的数据量调整计划，例如合并过小的分区、处理数据倾斜。

原始 Tick / 快照
      ↓
Spark 批作业：聚合、复权、历史回补
      ↓
质量校验
      ↓
日 K、分钟 K、因子表
      ↓
API 或其他数据服务

盘中实时结果可以先提供低延迟版本，盘后再用完整数据批量重算，并用最终结果覆盖临时结果。批作业应支持按日期范围运行，方便修复单日或一段历史。

from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder
         .master("local[4]")
         .appName("daily-bars")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

ticks = spark.read.parquet("data/ticks")
bars = (ticks.groupBy("code", "trade_date")
        .agg(F.min_by("price", "ts").alias("open"),
             F.max("price").alias("high"),
             F.min("price").alias("low"),
             F.max_by("price", "ts").alias("close"),
             F.sum("volume").alias("volume")))

bars.explain()
bars.write.mode("overwrite").partitionBy("trade_date").parquet("data/bars")

min_by("price", "ts") 按时间取最早成交价。分布式数据没有天然的行顺序，不能把 first("price") 当作开盘价。

如果数据按 trade_date 分区存储，下面的过滤可以触发分区裁剪：

ticks.filter("trade_date = '2026-10-28'").groupBy("code").count().explain()

执行计划中出现 PartitionFilters，说明其他日期的目录可以跳过；出现 Exchange，说明发生了 Shuffle。

1. 数据只有几 GB 也强行使用 Spark。pandas、Polars 或 DuckDB 在单机可处理时通常更简单、更快。
2. 以为写完 transformation 就已经计算。没有 action，计算通常还没有真正发生。
3. 忽略 Shuffle 分区数。数据很小时会产生大量空任务，数据很大时又可能单个任务过重。
4. 只按股票分组却不处理数据倾斜。个别活跃股票可能让少数任务远慢于其他任务。
5. 按日期和代码制造大量小文件。写出前要控制分区数，避免下游读取被小文件拖慢。
6. 重跑时直接追加写，导致结果重复。输出应按分区覆盖，或采用其他幂等写入方式。

• Spark 集群模式概览
• Spark RDD 编程指南
• Spark SQL 性能调优