如何在PySpark中提取每行集合列的最大值生成新列

• 场景：PySpark DataFrame的TAGID_LIST列每行存储一组数字，示例值为{426,427,428,430,432,433,434,437,439,447,448,450,453,460,469,469,469,469}
• 需求：逐行提取该组数字的最大值，上述示例预期返回469
• 初始实现代码：

wechat_userinfo.withColumn('TAG', f.when(wechat_userinfo['TAGID_LIST'] != 'null', max(wechat_userinfo['TAGID_LIST'])).otherwise('null'))

• 报错信息：TypeError: Column is not iterable

代码直接调用了Python内置的max()函数，这个函数仅支持处理Python原生可迭代对象（比如原生list、set），但传入的PySpark Column对象不属于Python原生可迭代结构，因此触发类型错误。
额外注意：PySpark自带的f.max()是跨行聚合函数，用来计算整列/分组内的最大值，不能用来逐行提取数组/集合内的元素最大值。

根据TAGID_LIST列的实际存储类型，二选一即可：

情况1：列是ArrayType/SetType数值类型（即存储的是原生数组/集合，不是带大括号的字符串） #

直接使用PySpark内置的行内数组处理函数f.array_max()即可，空值判断建议使用Column原生的isNotNull()方法，不要和字符串'null'做比对（除非列里存的真的是字符串值null）：

import pyspark.sql.functions as f

result = wechat_userinfo.withColumn(
    'TAG',
    f.when(
        f.col('TAGID_LIST').isNotNull(),
        f.array_max(f.col('TAGID_LIST'))
    ).otherwise(None)
)

如果列是SetType，f.array_max()可以直接识别处理，不需要额外做类型转换。

情况2：列是String类型（即存储的是带大括号的字符串，比如"{426,427,...}"格式的文本） #

需要先清洗字符串、拆分转成数值数组，再取最大值：

import pyspark.sql.functions as f

result = wechat_userinfo.withColumn(
    'TAG',
    f.when(
        f.col('TAGID_LIST') != 'null',
        f.array_max(
            f.split(f.regexp_replace('TAGID_LIST', r'[{}]', ''), ',')
            .cast('array<int>')
        )
    ).otherwise(None)
)

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