在Palantir Foundry转换装饰器中配置自定义检查的技术咨询

AI 小北

2026-8-14

解答

是的，你可以在 Palantir Foundry 的 @transform 装饰器中通过 Check API 配置自定义检查，支持使用 PySpark 内置函数或自定义 UDF 作为布尔条件。这样无需添加辅助列，就能直接在转换的检查结果中触发警告（或失败）。

数组唯一性检查的分步实现

1. 导入所需模块

首先从 Foundry 和 PySpark 导入必要的类：

from foundry.transforms import transform
from foundry.transforms.checks import Check
import pyspark.sql.functions as F

2. 定义自定义检查逻辑

你可以通过两种方式实现数组唯一性检查：使用 PySpark 内置函数（性能更优）或你已有的 UDF。

方案 A：使用 PySpark 内置函数（无需 UDF）

借助 array_distinct 和 size 函数避免使用 UDF，提升性能：

def array_has_unique_elements(col_name):
    return F.size(F.array_distinct(F.col(col_name))) == F.size(F.col(col_name))

方案 B：使用现有 UDF

如果你更倾向于使用自己的 UDF，保持原有注册方式即可：

@F.udf
def check_for_distinct_array_elements(arr):
    return len(set(arr)) == len(arr)

3. 在转换装饰器中应用检查

使用 Check.custom() 将自定义检查添加到转换中，指定检查条件、失败提示信息，以及触发警告（而非失败）的级别：

@transform(
    output=Output("/path/to/output"),
    input=Input("/path/to/input"),
    checks=[
        # 使用内置函数的检查
        Check.custom(
            condition=array_has_unique_elements("my_array_col1"),
            message="my_array_col1 包含重复元素",
            severity="WARNING"
        ),
        # 使用 UDF 的检查
        Check.custom(
            condition=check_for_distinct_array_elements(F.col("my_array_col2")),
            message="my_array_col2 包含重复元素",
            severity="WARNING"
        )
    ]
)
def my_transform(input):
    df = input.dataframe()
    df = (
        df
        .withColumn('my_array_col1', F.array(F.lit('first'), F.lit('second'), F.lit('third')))
        .withColumn('my_array_col2', F.array(F.lit('first'), F.lit('first')))
    )
    return df

关键提示

严重级别：使用 severity="WARNING" 仅触发警告而不终止转换，若需在检查失败时停止执行，可改为 "FAILURE"。
性能优化：PySpark 内置函数经过优化，在大数据集上的表现优于 UDF，建议优先使用。
检查可见性：这些检查的结果会显示在 Foundry 转换的「检查」标签页中，无需查看辅助列即可轻松监控和调试问题。

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