如何本地mock S3AFileSystem实现pytest测试spark.read.csv功能？

问题原因 #

报错核心是本地PySpark环境缺少Hadoop S3A文件系统的实现依赖，无法识别s3a:///s3://这类S3协议前缀，以下两种方案都可以在不上传真实文件到S3的前提下完成测试：

方案1：适配Mock S3的集成测试方案（测试真实读写逻辑） #

如果你需要验证Spark实际读取S3文件的完整流程，只需要修改test_spark_session fixture，在Spark启动时加载对应依赖并配置Mock S3的访问参数：

import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def test_spark_session():
    test_spark_session = (
        SparkSession.builder.master("local[*]")
        .appName("test")
        # 加载匹配Hadoop版本的S3A依赖，PySpark 3.1.2对应Hadoop 3.2版本
        .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:3.2.0,com.amazonaws:aws-java-sdk-bundle:1.11.375")
        # 配置S3A实现类
        .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
        # 指向moto mock S3的服务地址（默认端口为5000，自定义端口请自行修改）
        .config("spark.hadoop.fs.s3a.endpoint", "http://localhost:5000")
        # 关闭SSL校验、配置占位密钥、开启路径风格访问适配moto
        .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false")
        .config("spark.hadoop.fs.s3a.access.key", "test")
        .config("spark.hadoop.fs.s3a.secret.key", "test")
        .config("spark.hadoop.fs.s3a.path.style.access", "true")
        .getOrCreate()
    )
    return test_spark_session

修改后配合你已有的moto mock S3逻辑即可正常读取模拟上传的CSV文件。

方案2：轻量单元测试方案（仅验证函数本身逻辑，无额外依赖） #

如果你不需要测试Spark读S3的底层逻辑，只需要验证read_s3_csv_into_spark_df函数s3://转s3a://的逻辑是否正确，可以直接Mock掉Spark的读方法，不需要启动Mock S3服务：

from unittest.mock import MagicMock

def test_load_csv():
    # 构造mock的spark对象
    mock_spark = MagicMock()
    mock_csv_reader = MagicMock()
    mock_spark.read.csv = mock_csv_reader

    # 测试s3://前缀的转换逻辑
    test_uri = "s3://bucket/key/test.csv"
    read_s3_csv_into_spark_df(test_uri, mock_spark)
    # 验证传入csv方法的uri已经被替换为s3a前缀
    mock_csv_reader.assert_called_once_with("s3a://bucket/key/test.csv")

    # 可选：测试已经是s3a://前缀的场景
    mock_csv_reader.reset_mock()
    test_uri2 = "s3a://bucket/key/test.csv"
    read_s3_csv_into_spark_df(test_uri2, mock_spark)
    mock_csv_reader.assert_called_once_with("s3a://bucket/key/test.csv")

这种方案执行速度更快，不需要依赖外部服务，适合纯单元测试场景。

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