如何在PySpark中为re.findall()传入DataFrame字符串列？

你直接用Python标准库的re.findall()处理PySpark的Column对象会报错，因为re.findall()只能处理本地字符串，而F.col("Col_Name")是分布式的列对象，不是普通字符串。下面是两种可行的解决方法：

方法一：使用PySpark内置函数（推荐，Spark 3.1+支持） #

PySpark 3.1及以上版本提供了regexp_extract_all()函数，专门用于从字符串列中提取所有匹配正则捕获组的内容，返回数组类型的列：

from pyspark.sql import functions as F

# 示例：提取Col_Name列中所有数字
df = df.withColumn(
    "extracted_values",
    F.regexp_extract_all(F.col("Col_Name"), r"(\d+)", 1)
)

参数说明：

• 第一个参数：要处理的字符串列
• 第二个参数：带捕获组的正则表达式
• 第三个参数：捕获组的索引（从1开始，对应你要提取的分组）

方法二：自定义UDF（兼容旧版本Spark） #

如果你的Spark版本低于3.1，可以用自定义UDF封装re.findall()，将列中的每个字符串值传入处理：

import re
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StringType

# 定义处理函数，注意处理空值避免报错
def extract_matches(s):
    if not s:
        return []
    return re.findall(r"(\d+)", s)

# 注册UDF，指定返回类型为字符串数组
extract_udf = F.udf(extract_matches, ArrayType(StringType()))

# 应用到DataFrame
df = df.withColumn(
    "extracted_values",
    extract_udf(F.col("Col_Name"))
)

注意：内置函数的性能远优于UDF，因为UDF需要在Python虚拟机中处理数据，会产生额外的序列化开销，大数据量场景下优先用内置函数。

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