如何将PySpark DataFrame的行转换为自定义pouser类对象？

1. 修正pouser类的定义 #

你当前的类存在两个问题：一是@property装饰的方法缺少self参数，二是没有初始化方法来赋值私有属性。修正后的类如下：

class pouser:
    def __init__(self, name, max_val, avg_val, min_val=None):
        self.__name = name
        self.__max = max_val
        self.__avg = avg_val
        self.__min = min_val  # 你的DataFrame未提供Min列，这里设默认值

    @property
    def name(self):
        return self.__name

    @property
    def max(self):
        return self.__max

    @property
    def min(self):
        return self.__min

    @property
    def avg(self):
        return self.__avg

2. 处理DataFrame的重复列名 #

你的DataFrame包含两个Name列，会导致读取混乱，需要先重命名：

# 给重复列重命名，根据业务逻辑选择后续要用的Name列
df = df.toDF("Name", "Max", "DuplicateName", "Avg")

3. 转换为pouser对象列表 #

通过遍历DataFrame行来创建对象，若数据量不大可直接拉取到本地处理：

# 将Spark DataFrame转为本地行列表
rows = df.collect()

# 生成pouser对象列表
pouser_list = []
for row in rows:
    # 这里用第一个Name列赋值，Min列无数据传默认值None
    user = pouser(name=row["Name"], max_val=row["Max"], avg_val=row["Avg"])
    pouser_list.append(user)

验证转换结果 #

可以打印对象属性确认转换效果：

for user in pouser_list:
    print(f"Name: {user.name}, Max: {user.max}, Avg: {user.avg}, Min: {user.min}")

注意：如果DataFrame数据量极大，collect()会将全量数据拉到本地节点，可能引发内存溢出，这种情况建议用Spark的map算子做分布式转换。

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