palantir-foundry - Palantir Foundry 如何在计算中允许动态输入数量(代码库)

Question

我有一个文件夹，我会每月上传一个文件。该文件每个月的格式都相同。

第一个问题

想法是将此文件夹中的所有文件连接成一个文件。目前我正在对文件名进行硬编码(文件名[0]、文件名[1]、文件名[2]..)，但想象一下以后我将有 50 个文件，我应该明确地将它们添加到 transform_df 装饰器吗？有没有其他方法可以解决这个问题？

第二个问题:

目前我有 4 个文件(2021_07、2021_08、2021_09、2021_10)，我希望每当我添加显示 2021_12 数据的文件时避免更改代码。如果我添加 input_5 = Input(path_to_2021_12_do_not_exists) 代码将不会运行并给出错误。

如果不每月手动向我的代码添加一个新值，我如何为 future 的文件实现代码并让代码忽略不存在的输入？

谢谢

# from pyspark.sql import functions as F
from transforms.api import transform_df, Input, Output
from pyspark.sql.functions import to_date, year, col
from pyspark.sql.types import StringType
from myproject.datasets import utils
from pyspark.sql import DataFrame
from functools import reduce


input_dir = '/Company/Project_name/'
prefix_filename = 'DataInput1_'
suffixes = ['2021_07', '2021_08', '2021_09', '2021_10', '2021_11', '2021_12']

filenames = [input_dir + prefix_filename + suffixe for suffixe in suffixes]


@transform_df(
    Output("/Company/Project_name/Data/clean/File_concat"),
    input_1=Input(filenames[0]),
    input_2=Input(filenames[1]),
    input_3=Input(filenames[2]),
    input_4=Input(filenames[3]),
    )
def compute(input_1, input_2, input_3, input_4):
    input_dfs = [input_1, input_2, input_3, input_4]
    dfs = []

    def transformation_input(df):
        # some transformation
        return df
    for input_df in input_dfs:
        dfs.append(transformation_input(input_df))

    dfs = reduce(DataFrame.unionByName, dfs)
    return dfs

Answer 1

这个问题经常出现，简单的答案是你没有。定义数据集并在其上执行构建是在不同阶段执行的两个不同步骤。

每当您提交代码并运行检查时，您的整个 Python 代码都会在 renderSchrinkwrap 阶段执行，计算部分除外。这允许 Foundry 发现存在哪些数据集并发布。

发布涉及创建您的数据集，并将计算函数中的任何内容发布到数据集的作业规范中，因此类型转换厂知道在您运行构建时要执行的代码。

一旦您点击了数据集上的构建，Foundry 将只会选取作业规范上的任何内容并执行它。任何其他代码在您的检查期间已经运行，并且只运行了一次。

因此，任何动态输入/输出都需要您对您的存储库重新运行检查，这意味着必须进行一些代码更改，因为检查是 CI 过程的一部分，而不是构建的一部分。