PySpark RDD.map场景下部分函数mock不生效原因求解

问题底层原因 #

这个现象是Python mock的作用范围和PySpark的任务序列化执行机制共同导致的，核心逻辑如下：

1. PySpark RDD算子的闭包序列化规则 #

PySpark的RDD算子（比如map）属于惰性执行逻辑：你在调用transform_data系列函数时只是生成了执行计划，直到调用show()才会真正提交任务执行。提交任务时，Spark会把算子传入的函数以及它依赖的所有变量序列化，发送到Executor执行（哪怕是local[*]模式，执行逻辑也会经过序列化/反序列化步骤，和集群模式逻辑一致）。
序列化时的规则：

• 嵌套函数/lambda内部引用的局部变量、闭包变量，会随函数一起被序列化到任务包中，发送到Executor端直接使用。
• 模块顶层函数不会被完整序列化，仅序列化函数的导入路径，Executor端执行时会重新从磁盘加载对应模块、导入函数执行。

2. mock补丁的作用范围 #

你用patch('jobs.etl_job.get_random_bool')打的补丁，仅作用于Driver端当前内存中的jobs.etl_job模块对象，不会修改磁盘上的源码，也不会自动同步到Executor端重新加载的模块中。

3. 两类transform函数的差异 #

• transform_data1/2/3 生效的原因：三个函数里的get_random_bool都是被函数内部的lambda/嵌套函数直接捕获的闭包变量，你在patch上下文内调用transform函数时，jobs.etl_job.get_random_bool已经被替换为mock对象，这个mock对象会随lambda一起被序列化到任务中，Executor端执行时直接调用序列化过来的mock对象，所以返回预期的notabool。
• transform_data4/5 失效的原因：两个函数里的lambda仅引用了模块顶层函数f，序列化时只会把f的导入路径发送到Executor，Executor端执行时会重新从磁盘加载etl_job.py模块、导入f函数。此时Executor端的etl_job模块没有被打补丁，f调用的是原始的get_random_bool，所以返回真实的布尔值。

常见解决方案 #

• 方案1：单元测试时优先测试核心逻辑，把Spark RDD/DF的执行层抽离，不需要启动Spark就能测试转换逻辑。
• 方案2：如果需要在Spark执行逻辑中mock依赖，尽量把需要mock的对象直接捕获到算子的闭包中，避免通过顶层函数间接引用。
• 方案3：修改代码设计，把依赖通过参数显式注入到转换函数中，测试时直接传入mock对象即可。

内容的提问来源于stack exchange，提问作者offwhitelotus