如何在不同pytest测试函数间共享Spark DataFrame？

方案一：将DataFrame封装为Module级Fixture（推荐） #

把DataFrame的创建逻辑抽离成独立的pytest fixture，依赖已有的spark fixture并设置相同的module作用域，这样所有测试函数都能共享同一个DataFrame实例，同时保持测试的独立性。

修改后的完整代码：

import pytest
from pyspark.sql import SparkSession, Row
import os
import tempfile

@pytest.fixture(scope="module")
def spark() -> SparkSession:
    builder = SparkSession.builder \
        .appName('we_pipeline_test') \
        .config('spark.ui.showConsoleProgress', 'false') \
        .config('spark.sql.debug.maxToStringFields', '200')

    if 'SPARK_MASTER' in os.environ:
        builder = builder.master(os.environ['SPARK_MASTER'])
    else:
        builder = builder.master('local[1]')

    s = builder.getOrCreate()

    with tempfile.TemporaryDirectory("we_pipeline") as dir:
        s.sparkContext.setCheckpointDir(dir)
        yield s
        s.stop()

# 新增共享DataFrame的fixture，作用域与spark一致
@pytest.fixture(scope="module")
def shared_test_df(spark):
    # 替换为你实际的DataFrame创建逻辑
    data = [Row(id=1, name="Alice"), Row(id=2, name="Bob")]
    schema = "id INT, name STRING"
    df = spark.createDataFrame(data, schema)
    return df

def test_func1(spark, shared_test_df):
    # 直接使用共享的DataFrame进行测试
    assert shared_test_df.count() == 2

def test_func2(spark, shared_test_df):
    # 访问共享的DataFrame
    shared_test_df.show()

方案二：通过临时视图传递（不推荐） #

如果必须依赖test_func1的执行结果，可以通过Spark临时视图共享DataFrame，但这种方式会让测试耦合，依赖执行顺序，仅适合特殊场景：

def test_func1(spark, request):
    # 创建DataFrame
    data = [Row(id=1, name="Alice"), Row(id=2, name="Bob")]
    schema = "id INT, name STRING"
    df = spark.createDataFrame(data, schema)
    # 注册为临时视图
    df.createOrReplaceTempView("test_temp_view")
    # 标记测试已执行
    request.config.cache.set("test_func1_completed", "True")

def test_func2(spark, request):
    # 确保test_func1已执行
    assert request.config.cache.get("test_func1_completed", "False") == "True"
    # 从临时视图读取DataFrame
    df = spark.sql("SELECT * FROM test_temp_view")
    df.show()

总结 #

优先选择方案一，用fixture封装DataFrame，既保证了数据共享，又遵循了测试独立的原则，让代码更易维护。

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