3. Spark 基础回顾：核心概念与编程模型

在将 Backtrader 回测引擎与 Spark 集成之前，必须牢固掌握 Spark 的核心抽象与编程模型。本章从 RDD、DataFrame、Dataset 三大数据抽象出发，逐步深入到 Spark SQL 与 Structured Streaming，为后续的分布式回测引擎设计奠定基础。

3.1 三大核心数据抽象

Spark 提供了三种主要的数据抽象：RDD（弹性分布式数据集）、DataFrame 和 Dataset。它们分别面向不同的使用场景，在性能、类型安全和易用性上各有侧重。

特性 RDD DataFrame Dataset
引入版本 Spark 1.0 Spark 1.3 Spark 1.6
类型安全 强类型（编译时） 弱类型（运行时） 强类型（编译时）
数据表示 Java/Python 对象 Row 对象 + Schema JVM 类型对象
优化引擎 无（用户手动优化） Catalyst 优化器 Catalyst + Tungsten
适用场景 底层转换、非结构化数据 SQL 分析、ETL 类型安全 + 高性能
Python 支持 完整 完整 有限（PySpark 中为 DataFrame）
实践建议：在 Backtrader 与 Spark 集成场景中，优先使用 DataFrame 进行数据清洗与特征工程，使用 Dataset 进行类型安全的回测逻辑编写（Scala 环境），RDD 仅用于底层自定义分区或与旧版 API 兼容。

3.2 RDD：弹性分布式数据集

RDD 是 Spark 最基础的数据抽象，代表一个不可变、可分区、可并行计算的集合。RDD 的“弹性”体现在：当部分分区丢失时，可以通过 lineage（血统）重新计算。

3.2.1 RDD 的创建与操作

# 从集合创建
data = [("AAPL", 150.0), ("GOOG", 2800.0), ("MSFT", 300.0)]
rdd = spark.sparkContext.parallelize(data)

# 从文件创建
rdd_text = spark.sparkContext.textFile("hdfs://path/to/prices.csv")

# 转换操作（lazy）
filtered_rdd = rdd.filter(lambda x: x[1] > 200.0)
mapped_rdd = filtered_rdd.map(lambda x: (x[0], x[1] * 1.1))

# 行动操作（触发计算）
result = mapped_rdd.collect()
print(result)  # [('GOOG', 3080.0), ('MSFT', 330.0)]

3.2.2 RDD 的依赖关系

RDD 之间的依赖分为窄依赖（Narrow Dependency）和宽依赖（Wide Dependency），这决定了任务调度的效率与容错策略。

窄依赖：每个父 RDD 分区最多被子 RDD 的一个分区使用（如 map、filter）。
宽依赖：每个父 RDD 分区可能被子 RDD 的多个分区使用（如 groupByKey、reduceByKey），会产生 Shuffle。
RDD 窄依赖与宽依赖对比 RDD 依赖关系：窄依赖 vs 宽依赖 窄依赖（Narrow） P1 P2 P3 P1' P2' P3' 宽依赖（Wide） P1 P2 P3 P1' P2' P3' 窄依赖：无 Shuffle，可流水线执行，容错恢复快 宽依赖：产生 Shuffle，需要跨节点数据重分布，容错代价高

3.3 DataFrame 与 Dataset

DataFrame 是带 Schema 的分布式 Row 集合，类似于关系数据库中的表。Dataset 是 DataFrame 的类型安全版本，在 Scala/Java 中提供编译时类型检查。

3.3.1 DataFrame 的创建

# 从 SparkSession 创建
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("BacktraderSpark").getOrCreate()

# 从 RDD 转换
rdd = spark.sparkContext.parallelize([(1, "AAPL", 150.0), (2, "GOOG", 2800.0)])
df = rdd.toDF(["id", "symbol", "price"])

# 从 CSV 文件读取
df_prices = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("hdfs://path/to/historical_prices.csv")

# 从 Parquet 读取（推荐用于回测数据）
df_parquet = spark.read.parquet("hdfs://path/to/backtest_data.parquet")

3.3.2 DataFrame 的典型操作

# 查看 Schema
df_prices.printSchema()

# 选择与过滤
df_filtered = df_prices.select("symbol", "close_price") \
    .filter(df_prices.symbol == "AAPL")

# 聚合操作
df_avg = df_prices.groupBy("symbol") \
    .agg({"close_price": "avg", "volume": "sum"}) \
    .withColumnRenamed("avg(close_price)", "avg_price")

# 窗口函数（计算移动平均）
from pyspark.sql.window import Window
from pyspark.sql.functions import avg

window_spec = Window.partitionBy("symbol") \
    .orderBy("date") \
    .rowsBetween(-20, 0)

df_ma = df_prices.withColumn("ma_21", avg("close_price").over(window_spec))
性能注意：在回测数据预处理中，避免使用 UDF（用户自定义函数）处理大量行数据。优先使用内置 SQL 函数和窗口函数，它们经过 Catalyst 优化器深度优化，性能可提升 5-10 倍。

3.4 SparkSession：统一入口

SparkSession 是 Spark 2.0 引入的统一入口点，替代了旧的 SparkContext、SQLContext、HiveContext。在 Backtrader 与 Spark 集成中，所有操作都从 SparkSession 开始。

# 创建 SparkSession（推荐配置）
spark = SparkSession.builder \
    .appName("BacktraderSparkIntegration") \
    .config("spark.sql.shuffle.partitions", "200") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .getOrCreate()

# 获取 SparkContext（底层 API）
sc = spark.sparkContext

# 设置日志级别
spark.sparkContext.setLogLevel("WARN")

3.5 Spark SQL：结构化查询

Spark SQL 允许使用 SQL 语句直接查询 DataFrame，并自动应用 Catalyst 优化器。对于熟悉 SQL 的量化分析师，这是最直观的数据处理方式。

3.5.1 临时视图与 SQL 查询

# 注册临时视图
df_prices.createOrReplaceTempView("prices")

# 使用 SQL 查询
df_result = spark.sql("""
    SELECT 
        symbol,
        date,
        close_price,
        AVG(close_price) OVER (
            PARTITION BY symbol 
            ORDER BY date 
            ROWS BETWEEN 20 PRECEDING AND CURRENT ROW
        ) AS ma_21,
        STDDEV(close_price) OVER (
            PARTITION BY symbol 
            ORDER BY date 
            ROWS BETWEEN 20 PRECEDING AND CURRENT ROW
        ) AS volatility_21
    FROM prices
    WHERE date >= '2023-01-01'
""")

df_result.show(10)

3.5.2 Catalyst 优化器

Catalyst 是 Spark SQL 的查询优化器，它通过四个阶段将 SQL 或 DataFrame 操作转换为高效的物理执行计划：

  1. 分析（Analysis）：解析 SQL 语法树，绑定列名与数据类型
  2. 逻辑优化（Logical Optimization）：谓词下推、列剪枝、常量折叠
  3. 物理计划（Physical Planning）：选择 Join 策略、数据源扫描方式
  4. 代码生成（Code Generation）：使用 Tungsten 生成高效 Java 字节码
Catalyst 优化器工作流程 Catalyst 优化器：SQL → 物理执行计划 SQL 解析 （未解析逻辑计划） 分析 （解析逻辑计划） 逻辑优化 （优化逻辑计划） 物理计划 （选定物理计划） 代码生成 优化示例：谓词下推（Filter Pushdown）将 WHERE 条件提前到数据源扫描阶段 列剪枝（Column Pruning）只读取 SELECT 中涉及的列，减少 I/O

3.6 Spark Streaming 基础

Spark Streaming 是 Spark 的流处理引擎，支持微批处理（Micro-Batch）和连续处理（Continuous Processing）两种模式。在实时回测场景中，可以使用 Structured Streaming 处理实时行情数据。

3.6.1 Structured Streaming 核心概念

概念 说明 回测场景示例
Input Source 数据源（Kafka、文件、Socket） 从 Kafka 读取实时行情
Streaming DataFrame 无界表，持续追加新数据 实时 OHLCV 数据流
Trigger 触发微批处理的时间间隔 每 1 秒触发一次回测计算
Output Sink 结果输出（Console、File、Kafka） 输出实时信号到交易系统
Watermark 处理延迟数据的最大等待时间 允许行情延迟 5 秒

3.6.2 实时回测数据流示例

# 从 Kafka 读取实时行情
df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "market_data") \
    .load()

# 解析 JSON 格式的行情数据
from pyspark.sql.functions import from_json, col, to_timestamp

schema = "symbol STRING, price DOUBLE, volume LONG, timestamp STRING"

df_parsed = df_stream.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select(
    col("data.symbol"),
    col("data.price"),
    col("data.volume"),
    to_timestamp(col("data.timestamp")).alias("event_time")
)

# 添加 Watermark 并计算实时移动平均
df_ma_stream = df_parsed \
    .withWatermark("event_time", "10 seconds") \
    .groupBy(
        col("symbol"),
        window(col("event_time"), "1 minute", "30 seconds")
    ) \
    .agg({"price": "avg"}) \
    .withColumnRenamed("avg(price)", "ma_price")

# 输出到控制台（调试模式）
query = df_ma_stream.writeStream \
    .outputMode("append") \
    .format("console") \
    .trigger(processingTime="5 seconds") \
    .start()

query.awaitTermination()
集成要点：在 Backtrader 与 Spark 集成中，Structured Streaming 可用于实时回测场景。将 Backtrader 的 Cerebro 引擎封装为流处理函数，每个微批次触发一次回测迭代，实现准实时策略评估。注意设置合理的 Watermark 和 Trigger 间隔，避免回测延迟过高。

3.7 本章小结

本章回顾了 Spark 的核心概念：

  • RDD 是底层分布式数据集，适合自定义分区和容错逻辑
  • DataFrame 是带 Schema 的分布式表，适合 SQL 分析和 ETL
  • Dataset 提供类型安全，适合 Scala 环境下的复杂业务逻辑
  • SparkSession 是统一入口，管理所有 Spark 功能
  • Spark SQL 通过 Catalyst 优化器高效处理结构化数据
  • Structured Streaming 支持实时数据流处理，适用于实时回测
← 上一章 📖 返回目录 下一章 →