如何在Palantir Foundry代码库中使用数据集历史版本作为输入

Foundry的PySpark转换支持直接加载数据集的历史版本，无需依赖Contour的可视化操作，核心是通过transforms库的参数指定目标版本。

1. 通过版本ID加载历史版本 #

若已知目标版本的具体ID（可在数据集的「版本历史」页面获取），直接在Input中指定version参数：

from transforms.api import transform, Input, Output

@transform(
    output=Output("/your/output/dataset/path"),
    historical_data=Input("/your/input/dataset/path", version="v456")  # 替换为实际版本ID
)
def transform_historical_data(output, historical_data):
    # 读取历史版本数据集
    raw_df = historical_data.dataframe()
    
    # 执行自定义PySpark转换
    transformed_df = raw_df \
        .filter(raw_df["is_valid"] == True) \
        .withColumn("calculated_value", raw_df["base_value"] * 1.1)
    
    # 写入转换结果到输出数据集
    output.write_dataframe(transformed_df)

2. 通过时间戳加载历史版本 #

如果不清楚版本ID，可指定某个UTC时间点的快照，使用as_of参数（支持ISO 8601格式）：

from transforms.api import transform, Input, Output

@transform(
    output=Output("/your/output/dataset/path"),
    historical_data=Input("/your/input/dataset/path", as_of="2024-03-15T12:00:00Z")  # 替换为目标时间
)
def transform_historical_data(output, historical_data):
    raw_df = historical_data.dataframe()
    
    # 示例转换：分组统计
    aggregated_df = raw_df.groupBy("region").sum("sales_amount").withColumnRenamed("sum(sales_amount)", "total_sales")
    
    output.write_dataframe(aggregated_df)

关键注意事项 #

• 版本ID可在数据集详情页的版本列表中找到，格式可能是短标识（如v123）或完整UUID。
• as_of参数必须使用UTC时间，需与数据集版本的创建时间匹配。
• 若历史版本与当前代码依赖的schema存在差异，需在转换逻辑中添加字段兼容处理（如coalesce或select指定字段）。

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