多PySpark ETL流水线测试中全局RUN ENV环境变量冲突问题求助

针对你遇到的CI中多流水线测试因全局RUN ENV文件被占用导致的变量冲突问题，这里有几个优雅的解决方案：

方案1：重构RUN ENV为懒加载模式，配合fixture清理缓存 #

把原来模块导入阶段就加载RUN ENV的逻辑改成懒加载，只有在实际使用环境变量时才读取文件。这样测试fixture可以在每个测试执行前更新RUN ENV文件，然后清除缓存让后续代码读取最新值。

• 代码调整示例：

  # 替换原来的全局RUN_ENV = load_run_env_file()
  def get_run_env():
      # 用函数属性做缓存，避免重复读取文件
      if not hasattr(get_run_env, "_cached"):
          get_run_env._cached = load_run_env_file()
      return get_run_env._cached
  
  # 业务代码中使用时调用函数获取变量
  source_path = get_run_env().get("SOURCE_PATH")
  

• 配套pytest fixture：

  @pytest.fixture(scope="function")
  def setup_pipeline_env(request):
      # 获取当前测试对应的流水线环境配置
      pipeline_env_config = request.param
      # 写入RUN ENV文件
      write_run_env_file(pipeline_env_config)
      # 清除缓存，确保下一次get_run_env读取最新内容
      if hasattr(get_run_env, "_cached"):
          delattr(get_run_env, "_cached")
      yield
      # 测试结束后可以恢复默认环境（可选）
      write_run_env_file(default_env_config)
  

• 测试用例使用方式：

  @pytest.mark.parametrize("setup_pipeline_env", [pipeline1_env], indirect=True)
  def test_pipeline1_e2e(setup_pipeline_env):
      # 执行pipeline1的端到端测试
      ...
  

方案2：用临时文件隔离各流水线的RUN ENV #

不要用固定路径的全局RUN ENV文件，让代码支持通过环境变量指定RUN ENV路径。测试时为每个流水线分配独立的临时文件，完全隔离环境。

• 第一步：修改业务代码支持动态路径

  import os
  # 默认路径用原来的全局路径，同时支持环境变量覆盖
  RUN_ENV_PATH = os.getenv("RUN_ENV_PATH", "/path/to/global/run.env")
  RUN_ENV = load_run_env_file(RUN_ENV_PATH)
  

• 第二步：编写生成临时RUN ENV的fixture

  import tempfile
  import os
  @pytest.fixture(scope="module")
  def isolated_run_env(pipeline_config):
      # 创建临时文件写入当前流水线的环境配置
      with tempfile.NamedTemporaryFile(mode='w', delete=False) as temp_file:
          write_run_env_content(temp_file, pipeline_config)
      # 设置环境变量让业务代码读取这个临时文件
      os.environ["RUN_ENV_PATH"] = temp_file.name
      yield
      # 测试完成后清理临时文件和环境变量
      os.unlink(temp_file.name)
      if "RUN_ENV_PATH" in os.environ:
          del os.environ["RUN_ENV_PATH"]
  

• 第三步：为不同流水线测试绑定对应配置

  @pytest.mark.parametrize("isolated_run_env", [pipeline2_env], indirect=True)
  def test_pipeline2_e2e(isolated_run_env):
      # 执行pipeline2的端到端测试
      ...
  

方案3：解耦业务逻辑与RUN ENV依赖 #

把依赖RUN ENV的ETL代码改成参数注入模式，业务逻辑不直接读取全局变量，而是通过函数参数接收配置。测试时直接传入对应流水线的参数，完全绕开RUN ENV文件。

• 重构业务代码示例：

  # 原来的代码：直接依赖全局RUN_ENV
  def run_pipeline():
      source = RUN_ENV["SOURCE"]
      target = RUN_ENV["TARGET"]
      # PySpark ETL逻辑
      spark.read.parquet(source).write.parquet(target)
  
  # 重构后：依赖参数注入
  def run_pipeline(source_path: str, target_path: str):
      spark.read.parquet(source_path).write.parquet(target_path)
  

• 生产环境兼容：在生产入口脚本中读取RUN ENV传入参数

  if __name__ == "__main__":
      env = load_run_env_file()
      run_pipeline(
          source_path=env["SOURCE"],
          target_path=env["TARGET"]
      )
  

• 测试用例直接传参：

  def test_pipeline1():
      run_pipeline(
          source_path="s3://test/pipeline1/source",
          target_path="s3://test/pipeline1/target"
      )
      # 验证ETL结果
      ...
  

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