Python Spark中使用类对象调用transform函数报错，求排查方案

刚学习Python中的Spark专属函数，正在编写一个实现数据读取、转换、写入的程序，使用transform函数完成转换操作。但在调用transform函数时出现错误，以下是我的代码：

import logging
import sys
from datetime import datetime
from pyspark.sql import SparkSession


class Ingest:
    def __init__(self):
        self.spark_session = (
            SparkSession.builder.master("cluster")
            .appName("IngestData")  
            .getOrCreate()
        )
        self.input_directory = ("/mnt/input")
        self.output_directory = ("/mnt/output")

 

    def applyLogic(self):
        df.cache()

        listValues = ["X", "Y"]
        df = df.withColumn(
            "name",
            when(
                (df["Title"].contains("Gemini"))
                & (df["Title"].contains("Cancer")),
                "Intelligent"
            )
         return df




obj = Ingest()
transformedDF = obj.transform(applyLogic)

请问如何正确使用类对象调用transform函数？或是我的代码存在哪些问题？注：不使用transform函数时程序可正常运行。

你的代码存在多个关键问题，以下是逐一修正说明：

核心错误点 #

• DataFrame未定义：applyLogic方法中直接使用df，但没有从外部传入或在类中初始化这个DataFrame，且缺少数据读取步骤。
• transform调用方式错误：Spark的transform是DataFrame的方法，不是Ingest类对象的方法，需要先读取得到DataFrame后再调用。
• applyLogic方法参数缺失：作为transform的传入函数，需要接收DataFrame作为参数。
• 语法不完整：when函数未导入，且withColumn中的when缺少闭合括号，同时没有otherwise分支（否则不满足条件的行会显示null）。
• 缓存操作位置错误：应该在转换后或读取后缓存，且要基于传入的DataFrame操作。

修正后的完整代码 #

import logging
import sys
from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql.functions import when  # 导入when函数


class Ingest:
    def __init__(self):
        self.spark_session = (
            SparkSession.builder.master("cluster")
            .appName("IngestData")  
            .getOrCreate()
        )
        self.input_directory = "/mnt/input"
        self.output_directory = "/mnt/output"

    def read_data(self):
        # 读取数据，这里假设是parquet格式，可根据实际调整为csv、json等
        return self.spark_session.read.parquet(self.input_directory)

    def applyLogic(self, df):  # 接收DataFrame作为参数，符合transform的函数要求
        listValues = ["X", "Y"]  # 可根据实际需求使用该变量
        transformed_df = df.withColumn(
            "name",
            when(
                (df["Title"].contains("Gemini")) & (df["Title"].contains("Cancer")),
                "Intelligent"
            ).otherwise("Other")  # 添加otherwise分支，避免未匹配行出现null
        )
        transformed_df.cache()  # 对转换后的DataFrame缓存，减少重复计算
        return transformed_df

    def write_data(self, df):
        # 写入数据，这里假设是parquet格式，可根据实际调整
        df.write.mode("overwrite").parquet(self.output_directory)


# 执行完整ETL流程
obj = Ingest()
raw_df = obj.read_data()
transformedDF = raw_df.transform(obj.applyLogic)  # 对读取后的DataFrame调用transform
obj.write_data(transformedDF)

关键说明 #

1. transform的正确用法：df.transform(func)中，func必须是一个接收DataFrame并返回DataFrame的函数，这里直接传入类实例的applyLogic方法即可。
2. 代码结构优化：拆分出read_data和write_data方法，让ETL流程更清晰，便于维护。
3. 语法修复：补全when的导入和闭合括号，添加otherwise分支保证数据完整性。
4. 缓存合理使用：在转换完成后缓存DataFrame，避免重复执行转换逻辑。

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