在 Palantir Foundry 中进行测试答案

【问题标题】：Taking full Input dataset when testing transformations in Palantir Foundry在 Palantir Foundry 中进行测试
【发布时间】：2022-01-19 16:19:33
【问题描述】：

在 Palantir Foundry 中，我可以看到我们可以使用 Pytest 或 TransformRunner 编写单元测试。我的理解是，使用 Pytest，我们无法将转换的输出传递给单元测试，而在 TransformRunner 中，我们无法使用我们最初必须使用的数据集。我们需要一些测试数据。但是我想使用我的转换应该在生产中运行的整个输入数据集，并对它的输出进行测试。我怎样才能做到这一点？

【问题讨论】：

对于那些投票结束这个问题的人，如果你知道 Palantir Foundry 是如何工作的，这已经足够具体了，但我知道你来自哪里。 @DR_S 你的代表太低，看不到它，但你的问题已经有 2 票赞成“社区特定 - 需要调试信息”。为了避免堆栈溢出中的这些情况，我建议您使用一些您一直在尝试的代码示例来扩展您的问题。
要定义单元测试，您需要构建输入数据集以确保您将在稳定和恒定的数据上运行测试。实际上，使用生产数据运行测试并没有真正意义，因为这些数据可能会发生变化，您将无法预测（评估）结果（您应该能够在开发阶段运行测试，但是稍后如果进行任何更新以确保没有回归错误）。因此，您绝对需要考虑根据您的功能上下文构建测试用例并相应地构建测试数据集。

标签： unit-testing pyspark palantir-foundry

【解决方案1】：

您无法从 CI 访问铸造数据集，您需要将数据 sn-p 保存在您的存储库中的文件中，然后加载它。

test/fixtures/data/input/a.csv

col_a,col_b
1,2

TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'fixtures', 'data')


def test_runner_single_table(spark_session):
    pipeline = Pipeline()

    @transform_df(Output('/test_single_table/output/test'),
                  input_a=Input('/test_single_table/input/a'))
    def transform_1(input_a):
        return input_a.withColumn('col_c', input_a['col_a'] + input_a['col_b'])

    pipeline.add_transforms(transform_1)

    runner = TransformRunner(pipeline, '/test_single_table', TEST_DATA_DIR)

    output = runner.build_dataset(spark_session, '/test_single_table/output/test')
    assert output.first()['col_c'] == 3

TransformsRunner 会将Input 路径转换为目录路径。在上面的例子中：

TEST_DATA_DIR 告诉跑步者数据在您的环境中的位置
'/test_single_table' 告诉运行程序可以忽略哪些子路径，因为此路径仅存在于代工厂数据集上，而不存在于您的存储库中
input/a 将根据您在 repo 中定义的 Input('[ignored_sub_path]/input/a') 和文件夹结构进行解析。

如果您想更好地理解它们，您可以打印此属性，它会显示在 CI 检查中。

【讨论】：