你可以将dataframe传递给函数，见下面的代码。

def is_Data_Valid(df):
    df = df.withColumn(
        "ValidationErrors",
        f.when(
            f.col("Name").rlike("^[a-zA-Z]+$") & f.col("Age").cast("int").isNotNull() & f.col(
                "Experience").cast("int").isNotNull() & f.col("Year").cast("int").isNotNull() & f.col(
                "Dept").rlike("^[a-zA-Z]+$"),
            f.lit("0")
        ).otherwise(f.lit("Invalid data"))
    )

    return df

您可以通过传递所需的 Dataframe 来调用该函数，如下所示：

df_1_checked = is_Data_Valid(emp_df)
df_2_checked = is_Data_Valid(emp_1_df)

只需将DataFrame作为参数传入，如下所示：

from pyspark.sql import DataFrame

def is_Data_Valid(df: DataFrame) -> DataFrame:
    return df.withColumn(
        "ValidationErrors",
        f.when(
            f.col("Name").rlike("^[a-zA-Z]+$") & f.col("Age").cast("int").isNotNull() & f.col(
                "Experience").cast("int").isNotNull() & f.col("Year").cast("int").isNotNull() & f.col(
                "Dept").rlike("^[a-zA-Z]+$"),
            f.lit("0")
        ).otherwise(f.lit("Invalid data"))
    )

记住总是从这样的函数中返回DataFrame-PySpark函数不是 * 就地 * 执行的，而是每个DataFrame都是不可变的，所以无论何时执行任何转换，都必须创建一个新的示例。