Spark 批处理与结构化流测试

• Spark 本地模式
• RDD/DataFrame 转换逻辑验证
• 规避集群依赖

使用 Spark 本地模式（local[*]），在 JVM 内启动 SparkContext，无需真实集群即可对转换逻辑做单元测试。测试时用 SparkSession.builder().master("local[2]").appName("test").getOrCreate() 创建本地 SparkSession，用 SparkContext.parallelize 构造 RDD 或 createDataFrame 构造 DataFrame，对转换逻辑（map/filter/join/聚合）执行后 collect 结果并断言。用 spark.time() 或 funSuite、ScalaTest 组织用例。要规避集群依赖，可将数据源替换为内存集合、将外部依赖（HDFS、Kafka）mock 或替换为本地文件，并在 after 中关闭 SparkSession 释放资源。对小数据集做逻辑验证，保证转换正确后再上集群回归。

本地模式让 Spark 逻辑在测试环境内快速验证，反馈快、无需集群资源。核心是"逻辑与集群解耦"：用内存数据验证转换逻辑，用真实数据源做集成测试。这样既能快速迭代，又能避免集群资源竞争与调度不稳定的干扰。

• 事件时间与水位线
• 窗口计算
• 测试中的时间控制

在测试中通过构造带事件时间戳的输入数据，配合 withWatermark 设置水位线，可控地验证窗口计算与迟到的数据（late data）。核心是：用inputStream 维护输入数据，通过 advanceManualClock 或 MemoryStream 的 addData 手动推进时钟，控制事件时间与处理时间。测试时构造多个时间戳的数据（如 12:00、12:05、12:10），用 window(eventTime, "10 minutes") 定义窗口，设置 watermark（如 10 分钟），验证窗口的触发时机、窗口内聚合结果，以及迟到数据（超过 watermark 被丢弃、在 watermark 内被重新聚合）的行为。通过手动控制时钟与水位线，能精确断言窗口计算的触发与结果。

事件时间与水位线是流式计算正确性的关键。测试通过手动推进时钟和注入延迟数据，能确定性地验证窗口触发时机与迟到数据策略，避免依赖真实时钟导致的不确定性。这是流式窗口测试的核心手段。

• Catalyst 优化器
• 逻辑计划与物理计划
• 计划差异对结果的影响

Catalyst 优化器会把输入的 SQL 逻辑计划（LogicalPlan）经过分析、优化（谓词下推、列裁剪、常量折叠、join 重排）后生成物理计划（PhysicalPlan），再决定实际的执行算子与 shuffle。逻辑计划与物理计划的差异可能导致结果不同：例如 join 重排改变了 join 顺序进而影响 shuffle 的数据分布与 NULL 处理；谓词下推改变了过滤的执行位置；列裁剪消除了未使用列的计算。测试断言需注意：同一 SQL 在不同 Spark 版本或优化开关下，物理计划可能不同，导致浮点求和顺序、聚合顺序、结果行顺序不同。正确做法是断言结果集（排序后）而非执行顺序，并可通过 df.queryExecution().logical() 与 physical() 检查计划是否符合预期，必要时用 explain() 输出验证优化是否生效。

Catalyst 让"写法"与"执行"解耦，测试必须关注物理计划而非仅依赖 SQL 字面。过度依赖执行顺序的断言会因优化而失败（假失败），而忽略优化差异又会漏掉真实 bug。因此断言排序后的结果集并检查物理计划是关键。

• 数据倾斜的识别
• 修复方案（加盐/广播/重分区）
• 修复效果验证

测试倾斜修复方案是否生效，需先识别倾斜（抽样统计 key 分布，找出热点 key 与 Task 处理时间差异），然后分别验证三种修复方案。加盐（salt）：给热点 key 加随机后缀打散，验证打散后各 Task 数据量均衡、耗时下降，且结果与未加盐一致。广播（broadcast）：对小表广播避免 shuffle，验证 join 中无 shuffle、Task 数减少、耗时下降。重分区（coalesce/repartition）：调整分区数，验证分区数据分布均衡、无热点。验证手段包括：比对修复前后任务的 stage 耗时、Task 最大/平均耗时、shuffle 数据量、关键 key 在不同 Task 的分布，以及最终结果正确性（与未修复基线一致）。只有在"结果正确 + 耗时/均衡度改善"时才算修复生效。

倾斜修复的核心风险是"修了性能却改了结果"。因此测试必须同时断言"结果一致"与"性能改善"，否则修复可能引入正确性 bug。用耗时分布与数据分布量化验证，能客观判断修复是否真正生效。

• 失败重试与 Checkpoint
• 幂等性验证
• 输出一致性与重复

验证幂等性的核心是：同一输入在任务失败重试或从 Checkpoint 恢复后，输出数据与首次成功执行一致，无重复、无丢失。方法包括：对同一输入执行多次（含失败重试），比对输出结果（行数、内容、主键唯一性）是否一致；对输出做幂等去重校验，确认同一主键在输出中不重复；通过 Checkpoint 恢复后，验证从 checkpoint 处继续执行不重复处理已处理的数据。对流式场景，验证 sink 的 Exactly-Once 输出（如用 Kafka 的事务、或基于主键的幂等写入）。测试可在执行中途 kill 任务再恢复，比较恢复后输出与完整执行输出的差异。

幂等性是保证批流任务可靠重跑与恢复的根本。测试通过"重试 + 恢复 + 对账"验证输出不重复不丢失，能暴露 checkpoint 设计缺陷与 sink 重复写入问题，是数据一致性保障的关键用例。

• 谓词下推
• 分区裁剪
• 执行计划级断言

通过检查 Spark 的物理计划（df.queryExecution().executedPlan）断言谓词下推与分区裁剪是否生效。谓词下推：断言 Filter 算子是否出现在 Scan 之前（下推到数据源，如 Parquet 的 row group 过滤、JDBC 的 WHERE 下推），确认过滤条件被下推到数据源层减少扫描量。分区裁剪：断言 Scan 的 PartitionFilters 包含预期的分区条件，确认只扫描目标分区而非常表扫描。实现上可解析 executedPlan 的字符串表示（plan.toString()）或遍历算子树，检查是否包含预期算子（如 FileScan 的 pushedFilters、partitionFilters）。测试断言计划中过滤/裁剪生效，可防止 SQL 写成全表扫描导致性能回归。

执行计划级断言能验证"优化是否生效"，而不只是结果正确。谓词下推未生效会导致全表扫描的性能问题；分区裁剪未生效会导致扫描全部分区。通过断言计划中的算子与过滤条件，能提前发现性能风险。

• RDD/DataFrame 单元测试
• 多数据源集成测试
• 测试分层

单元测试在本地模式验证转换逻辑，用 parallelize/createDataFrame 构造输入，断言转换结果。集成测试验证与真实数据源的对接：Hive 用 Hive 元数据与 Hive 表做读写集成测试，验证 SQL 在 Hive 表上的执行与分区；Kafka 用嵌入式 Kafka（如零依赖的 in-memory broker）或真实测试集群做读写集成测试，验证消费/生产、序列化、offset 管理。集成测试要覆盖真实数据源的 schema、类型、分区、序列化格式，确保单元测试验证的逻辑在真实环境下正确。测试分层：单元测试（逻辑）→ 数据源集成测试（对接）→ 端到端测试（完整链路）。

单元测试保证逻辑正确，集成测试保证对接正确。真实数据源（Hive/Kafka）的 schema、序列化、分区等细节无法在单元测试中覆盖，因此必须做集成测试。分层的测试策略兼顾速度与真实性。

• 水印与窗口
• 延迟数据处理
• 流式场景构造

用 MemoryStream 或手动推进时钟的方式构造流式场景。测试水印：设置 withWatermark 后注入不同事件时间的数据，验证超过 watermark 的迟到数据被丢弃、未超过的迟到数据被重新聚合。测试窗口：定义不同窗口（如 5 分钟/10 分钟 tumbling 或 sliding），注入跨窗口边界的数据，验证窗口触发与聚合结果。测试延迟数据处理：注入晚于 watermark 的迟到数据，验证其被丢弃且不改变已输出结果；再注入晚到但在 watermark 内的数据，验证其被重新聚合并触发更新输出。通过控制事件时间与时钟，确定性验证每个场景的行为。

流式测试的关键是"确定性控制时间"。通过手动推进时钟与注入特定时间戳数据，能精确验证水印、窗口与迟到数据策略，避免真实时钟带来的不确定性。这是结构化流测试的标准做法。

• Exactly-Once 语义
• Sink 重复写入
• 幂等输出验证

Exactly-Once 要求每条数据被消费且仅被处理并输出一次。验证方法：模拟 Sink 重复写入场景（如 Sink 因故障重试导致同一批次被写两次），用 Kafka 事务或幂等 Sink（基于业务主键去重）保证最终只落一份。测试时构造重复批次输入，验证输出后主键无重复、总量正确。可设计唯一键（如事件 ID）做幂等校验：对输出做 groupBy uid count，断言无 count>1 的键。也可通过用户提供批次号的 Sink 实现（ForeachWriter 的 open 回调携带批次号 epochId），验证同一批次重复提交时数据被幂等覆盖。最终断言输出与"只处理一次"的期望结果一致。

幂等输出是流式 Exactly-Once 落地的关键。通过唯一键去重与批次幂等写入，能从"结果"层面保证不重复不丢失，而不依赖底层事务机制。测试验证了幂等写作在重复写入下的最终一致性。

• 闭包序列化
• 未序列化对象与外部状态
• 广播变量

Spark 会把闭包（lambda、UDF 引用的对象）序列化后分发到 executor，若闭包引用了不可序列化对象（如连接、非 transient 的类实例）会在任务提交时抛 NotSerializableException。外部状态（如闭包内可变全局变量、非序列化 context）会导致序列化失败或结果不确定。提前暴露的方法：在单元测试中显式序列化闭包（new ObjectOutputStream 序列化引用的对象），或用 SparkContext 本地模式触发序列化检查；将外部状态改用广播变量（broadcast）传递，广播变量只序列化一次且只读，避免闭包捕获大对象。测试应断言任务能正常序列化并执行，且对每次 executor 结果一致。可开启 spark.serializer 与序列化检查，或在测试中刻意引用不可序列化对象验证报错路径。

序列化陷阱是 Spark 分布式任务最常见的运行期失败。提前在测试中序列化闭包、用广播变量替代外部状态，能暴露并规避问题。理解"闭包在 executor 分布执行"的模型，是设计此类测试的前提。

• repartition 与 coalesce 差异
• shuffle 行为
• 数据分布与并行度断言

repartition 会触发全量 shuffle（打散数据到指定分区数），coalesce 在上游分区数大于目标时避免 shuffle（仅合并分区，减少 shuffle）。测试断言：用 rdd.getNumPartitions() 验证分区数变化；用 rdd.mapPartitions(计数) 或 rdd.glom() 统计各分区数据分布，验证 shuffle 后数据是否均匀、coalesce 后分区是否保留数据；通过 rdd.toDebugString 或 DAG 判断是否发生 shuffle。并行度变化：设置 spark.sql.shuffle.partitions 或 repartition 的并行度，验证并行度对 Task 数与执行时间的影响。断言时应关注"结果正确性不因分区数/并行度变化而改变"（如聚合、排序在分区变化后仍一致），同时验证数据分布是否均衡。

repartition 与 coalesce 影响 shuffle 与执行效率，但不改变逻辑结果（除非依赖分区内部顺序）。测试需区分"逻辑正确性"与"物理分布"，断言结果一致的同时验证 shuffle 行为与分布均衡，防止性能问题。

• AQE 机制
• 查询计划动态调整
• 测试稳定性

AQE（Adaptive Query Execution）会在运行时根据实际数据统计动态调整查询计划（如动态合并 shuffle 分区、动态切换 join 策略、动态调整 join 顺序），导致不同数据规模下执行计划不同，进而使测试结果（如分区数、Task 数、结果行顺序）不稳定。防止方法：断言时只关注逻辑结果（排序后的数据），不依赖分区数、Task 数、执行计划等物理细节；对涉及 AQE 的测试，固定数据规模或固定 spark.sql.adaptive.coalescePartitions.enabled 等参数，使计划可复现；对 AQE 的功能性测试（如动态 join 选择）单独验证其行为，而非依赖其结果一致性。必要时用 explain 记录 AQE 调整后的计划用于诊断。

AQE 的"自适应"本质是让计划随数据变化，这天然与"固定计划断言"冲突。测试应把逻辑结果与物理计划解耦，固定参数或忽略物理细节，才能保证不同规模下稳定。理解 AQE 的调整点是设计稳定测试的关键。

• 数据倾斜验证
• shuffle 调优
• OOM 场景定位

数据倾斜验证：统计 key 分布与各 Task 处理时间，定位热点 key 与耗时差异，通过采样、加盐、广播等修复后对比耗时与分布。shuffle 调优验证：对比不同 spark.shuffle.partitions、spark.sql.shuffle.partitions、批次大小等参数下的耗时、shuffle 数据量与 Task 数，确认调优效果。OOM 场景：通过构造大内存数据（聚合大 key、join 大表、cache 大量数据）触发 OOM，观察 executor 内存使用、GC 与失败日志，定位是内存不足、数据倾斜还是 driver 端 OOM，通过增大内存、分区、避免 collect 全量数据等优化。稳定性测试：在数据量、并发、资源变化下反复运行，验证任务不 OOM、不失败、结果稳定。定位手段包括 Spark UI、日志、executor 内存监控。

性能与稳定性问题（倾斜、shuffle、OOM）相互关联，需量化定位。通过构造场景触发问题、对比参数与分布、监控资源，能定位根因并验证优化。稳定性测试保证了在数据与资源波动下任务可靠。

• 模拟数据生成
• 数据回放
• 适用场景

模拟数据：按业务规则与分布随机生成测试数据，覆盖边界、异常、倾斜等场景，适用于逻辑验证与新的功能测试，可精确控制数据形态。数据回放：抓取生产数据（或生产流式数据的 trace）回放到测试环境，保留真实分布、关联关系与时间序列，适用于回归测试、性能与稳定性测试，能真实反映生产环境的行为。回放需要脱敏与截取（按时间窗口/比例抽样），并注意与生产数据的一致性。两者结合：模拟用于快速覆盖与边界，回放用于真实性与回归。对流式场景，回放可模拟真实事件流的时间序列与分布。

模拟数据可控但可能与生产偏离，回放数据真实但可控性差。正确选择取决于目标：验证新逻辑用模拟，验证生产兼容与回归用回放。两者互补是数据测试的最佳实践。

• 输出验证
• 血缘验证
• 数据一致性

输出验证：对 Spark 任务的输出做行数、内容、聚合值、主键唯一性的对账，与期望基线（golden）或上游数据比对，确认结果正确、无丢失无重复。血缘验证：通过 Spark 的 RDD 血缘（lineage）或 SQL 血缘（解析 SQL 得 DAG），确认输出表的上下游依赖关系正确，验证变更影响范围与下游消费。血缘测试可断言：输出表的数据确实来自预期的上游表；上游变更时下游能正确感知；任务重跑时血缘关系不变。输出与血缘结合，能验证"数据从哪来、是否正确到达"。

输出验证保证结果的正确性，血缘验证保证数据来源与依赖关系的正确性。二者结合既能确认"结果对"，又能确认"来源对"，是数据任务完整验证的两个维度。

• 处理延迟
• 事件时间延迟
• 延迟度量与断言

处理延迟是记录从进入流到被处理完的耗时（不包含事件时间跨度），度量方法是在处理逻辑中记录输入时间戳与完成时间戳，或在接收端记录批次到达与处理完成时间，统计平均/最大延迟。事件时间延迟是数据产生（事件时间）到被处理（处理时间）的差值，反映端到端时效，度量方法是对每个事件记录 eventTime 与 processingTime 的差值。应设置触发器（如 trigger(ProcessingTime(10s))）控制批次粒度。断言：通过观察延迟的分布（p50/p95/p99）与阈值（如 SLA），验证处理延迟与事件时间延迟不超过预期；监控背压与积压延迟，确认系统在负载下延迟可控。可构造固定速率输入测量稳定的延迟基线。

端到端延迟是流式系统的核心 SLA。区分"处理延迟"与"事件时间延迟"能定位延迟来源（计算慢 vs 数据迟到）。通过统计延迟分布与阈值断言，能验证延迟是否满足业务要求。

• 种子数据管理
• golden 文件
• 版本化与更新

种子数据（输入基线）与 golden 文件（期望输出）应纳入版本管理，与代码一起提交。管理方法：种子数据存储为固定格式（CSV/JSON/Parquet）并命名带版本，golden 文件与测试用例一一对应，记录生成者与时间。更新流程：当逻辑变更导致预期输出变化时，先人工审查变更合理性，确认后重新生成 golden 文件并更新版本，禁止直接覆盖旧 golden 导致回归失效。采用"审查 + 生成 + 版本化"流程，可通过 golden 文件的 diff 审查变更。测试运行时加载当前版本的种子与 golden，与输出比对，若有差异则指出是代码 bug 还是 golden 过期。可用 golden 文件的历史版本做回归对比。

golden 文件是回归测试的基线，其版本管理直接决定回归可信度。随意覆盖 golden 会掩盖 bug。通过"审查变更 + 版本化 + 生成流程"能保证基线可信且可追溯。

• Spark 2/3 行为差异
• 兼容性回归
• 结果一致性

Spark 2 与 3 在 SQL 解析、优化器（Catalyst 升级、AQE 引入）、默认行为（如 SQL 方言、spark.sql.legacy 开关、ANSI 模式）上存在差异，可能导致同一任务结果不同。回归方法是：在 Spark 2 与 3 环境分别执行同一任务，对结果做对账（行数、聚合值、排序后内容），定位差异点。重点验证：类型转换的默认行为（如字符串转数字、日期解析）、NULL 与空串处理、窗口函数边界、legacy 参数开关、AQE 开关对结果的影响。对已知差异设置兼容开关（如 spark.sql.legacy.parquet.datetimeRebaseMode）或改写 SQL，使其在双版本结果一致。回归时建立双版本结果基线，升级时触发全量回归。

Spark 升级是高风险变更，行为差异可能静默改变结果。通过双版本对账与兼容开关，能识别并处理差异，防止升级导致数据质量事故。回归策略是升级前的必要保障。

• cache 与 persist 差异
• 存储级别选择
• 内存降级与失效

cache 默认用 MEMORY_ONLY 级别，persist 可指定存储级别（MEMORY_AND_DISK、MEMORY_ONLY_SER、DISK_ONLY 等）。验证存储级别选择：断言 persist 后 rdd.getStorageLevel 符合预期，比较不同级别下缓存命中率与重复计算耗时。内存不足降级：当数据超过内存时，MEMORY_ONLY 会丢失部分缓存（重新计算），MEMORY_AND_DISK 会溢出到磁盘，测试构造大数据集验证降级行为、是否丢数据（结果仍正确）与性能差异。失效时机：验证 unpersist() 后缓存被清除、Storage tab 中缓存消失、重新计算被触发；验证 checkpoint 与 persist 的配合。断言缓存命中能减少重复计算耗时，且无论降级与否结果一致。

缓存策略影响性能与内存。测试需验证存储级别、降级行为与失效时机，确保缓存正确且不因内存不足导致结果错误或性能恶化。缓存"失效"与"降级"是与"结果正确"解耦的物理行为，需独立验证。