PySpark代码Pytest单元测试遇HADOOP_HOME未设置错误求助

问题描述 #

使用Pytest对PySpark代码进行单元测试时，预期仅依赖pyspark包即可完成测试（无需Spark/Hadoop运行时，如Jenkins环境），但测试过程中出现两个问题：

1. 抛出java.io.FileNotFoundException，提示HADOOP_HOME和hadoop.home.dir未设置；
2. 测试断言失败（AssertionError）。

业务代码片段 #

def read_inputfile_from_ADLS(self):
    try:
        if self.segment == "US":
            if self.input_path_2 is None or self.input_path_2 == "":
                df = self.spark.read.format("delta").load(self.input_path)
            else:
                df = self.spark.read.format("delta").load(self.input_path_2)
    except Exception as e: 
        resultmsg = "error reading input file"

测试代码 #

import pytest
from unittest.mock import patch,MagicMock , Mock

class TestInputPreprocessor:
    inpprcr = None
    dataframe_reader = 'pyspark.sql.readwriter.DataFrameReader'

    def test_read_inputfile_from_ADLS(self,spark,tmp_path):
        self.segment = 'US'
        self.input_path_2 = tmp_path 
        with patch(f'{self.dataframe_reader}.format', MagicMock(autospec=True)) as mock_adls_read:
            self.inpprcr.read_inputfile_from_ADLS()
            assert mock_adls_read.call_count == 1

错误日志 #

AssertionError
---------------------------------------------- Captured stderr setup ------------------- 
--------------------------- 
23/07/12 23:58:42 WARN Shell: Did not find winutils.exe: java.io.FileNotFoundException: 
java.io.FileNotFoundException: HADOOP_HOME and hadoop.home.dir are unset. -see 
https://wiki.apache.org/hadoop/WindowsProblems
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use 
setLogLevel(newLevel).
23/07/12 23:58:42 WARN NativeCodeLoader: Unable to load native-hadoop library for your 
platform... using builtin-java classes where applicable

解决方案 #

1. 消除HADOOP_HOME未设置警告 #

该警告不会中断测试，但会干扰日志输出。可通过以下方式抑制：

• 在测试代码中设置临时环境变量：
  在测试文件开头添加：

  import os
  # 设置HADOOP_HOME为任意非空路径（无需实际Hadoop安装）
  os.environ['HADOOP_HOME'] = '.'
  # 抑制不必要的驱动类路径检查
  os.environ['PYSPARK_SUBMIT_ARGS'] = '--driver-class-path /dev/null pyspark-shell'
  

• Jenkins环境配置：在构建任务中添加环境变量HADOOP_HOME，值设为当前目录（.）即可。

2. 修复断言失败问题 #

原测试代码存在两处核心问题：

• 被测实例self.inpprcr未初始化，调用方法会抛出AttributeError；
• 补丁目标错误：直接patchDataFrameReader类的format方法，而非实例对象的方法，导致无法捕获实际调用。

修改后的测试代码示例：

import pytest
from unittest.mock import MagicMock
from your_module import InputPreprocessor  # 替换为实际模块名

class TestInputPreprocessor:
    def test_read_inputfile_from_ADLS(self, spark, tmp_path):
        # 1. 初始化被测实例并注入必要属性
        inpprcr = InputPreprocessor()
        inpprcr.spark = spark
        inpprcr.segment = 'US'
        inpprcr.input_path_2 = tmp_path

        # 2. Mock spark.read对象及其方法链
        mock_read = MagicMock()
        inpprcr.spark.read = mock_read

        # 3. 执行被测方法
        inpprcr.read_inputfile_from_ADLS()

        # 4. 断言方法调用符合预期
        # 验证format被传入"delta"参数调用一次
        mock_read.format.assert_called_once_with("delta")
        # 验证load被传入input_path_2的值调用一次
        mock_read.format.return_value.load.assert_called_once_with(tmp_path)

关键修改说明

• 正确初始化InputPreprocessor实例，并设置业务逻辑依赖的spark、segment、input_path_2属性；
• 直接mockspark.read实例对象，而非类方法，确保捕获实际代码中的调用链；
• 断言更贴合业务逻辑：不仅验证调用次数，还验证传入的参数是否正确，提升测试的准确性。

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