如何筛选Spark DataFrame中版本高于指定值的记录？

问题背景 #

原始Spark DataFrame定义如下：

df = spark.createDataFrame([[1, '3.1.5'],
                            [2, '1.23.0'],
                            [3, '0.2.0'],
                            [4, None]], ['Row', 'Version'])

需求是筛选出Version列值高于指定版本（如v='1.2.0'）的记录，预期结果：

+---+-------+
|Row|Version|
+---+-------+
|  1|  3.1.5|
|  2| 1.23.0|
+---+-------+

尝试使用packaging库编写UDF实现时，出现以下异常：

AttributeError: 'Version' object has no attribute '_get_object_id'

原尝试代码：

from packaging import version
from pyspark.sql import functions as F
from pyspark.sql import types as T

def version_parse(text):
    try:
        vers = version.parse(text)
    except TypeError:
        vers = None        
    return str(vers)
version_parse_udf = F.udf(version_parse, T.StringType())

df = df.filter(version_parse_udf(F.col('Version')) > version.parse('1.2.0')) 

错误原因 #

原代码的核心问题是：

• version_parse_udf(F.col('Version'))返回的是Spark的String类型列，而version.parse('1.2.0')是Python的Version对象
• Spark无法识别Python的Version类型，在尝试比较Spark列和Python对象时，抛出了属性不存在的错误
• 另外，将Version对象转成字符串后再比较，会存在版本号语义错误（比如字符串'1.10.0'会被认为小于'1.2.0'）

解决方案 #

方案一：在UDF内部完成版本比较（依赖packaging库） #

修改UDF逻辑，直接在UDF中完成版本解析与比较，返回布尔值供filter使用：

from packaging import version
from pyspark.sql import functions as F
from pyspark.sql import types as T

def is_version_higher(text, target_version):
    if text is None:
        return False
    try:
        vers = version.parse(text)
        return vers > target_version
    except:
        # 处理无效版本号的情况
        return False

# 预解析目标版本，传入UDF
target_vers = version.parse('1.2.0')
version_compare_udf = F.udf(lambda x: is_version_higher(x, target_vers), T.BooleanType())

# 执行过滤并展示结果
df.filter(version_compare_udf(F.col('Version'))).show()

方案二：无外部依赖，拆分版本号为整数数组比较 #

将版本号拆分为整数数组，利用Spark的数组比较特性（按元素顺序依次比较，符合版本号语义）：

from pyspark.sql import functions as F
from pyspark.sql import types as T

def split_version(version_str):
    if version_str is None:
        return []
    parts = version_str.split('.')
    try:
        # 拆分后转成整数数组
        return [int(p) for p in parts]
    except:
        # 无效版本号返回空数组，会被过滤掉
        return []

split_version_udf = F.udf(split_version, T.ArrayType(T.IntegerType()))

target_version = [1, 2, 0]
# 添加版本数组列，过滤后删除临时列
df.withColumn('version_parts', split_version_udf(F.col('Version')))\
  .filter(F.col('version_parts') > target_version)\
  .drop('version_parts')\
  .show()

两种方案说明 #

• 方案一依赖packaging库，能处理更复杂的版本号格式（如带预发布标签的版本），语义更准确
• 方案二无需外部依赖，适用于仅需处理标准三段式版本号的场景，轻量且兼容性好

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