基于PySpark DataFrame利用正则提取动态参数名与对应值

我们有一个存储字符串值的Spark DataFrame，需要通过DataFrame转换提取{}中指定的参数名及其对应值，参数名称不固定，每行可能包含不同参数。具体示例如下：

创建该DataFrame的代码如下：

from pyspark.sql.types import StructType, StructField, StringType

df = spark.createDataFrame([
    ("https://{accountName}.queue.core.windows.net/?restype=service&timeout={timeout}", "https://stackoverflow.queue.core.windows.net/?restype=service&timeout=20"),
    ("https://{accountName}.queue.core.windows.net/?restype=service&timeout", "https://customer-academy.queue.core.windows.net/?restype=service")
    ], 
    StructType([StructField("pattern", StringType(), True), StructField("sample_input", StringType(), True)])
)

可以通过**自定义UDF（用户定义函数）**实现动态参数提取，核心思路是根据每行的pattern生成匹配规则，再从sample_input中捕获对应参数值，具体步骤如下：

1. 导入依赖库 #

import re
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType, MapType

2. 定义参数提取函数 #

这里提供两种返回格式的实现：

方式一：返回指定格式的字符串

def extract_params_str(pattern, sample_input):
    # 提取pattern中所有被{}包裹的参数名
    params = re.findall(r'\{(.*?)\}', pattern)
    if not params:
        return "{}"
    
    # 将pattern转换为正则匹配模板，把{param}替换为捕获组(.*?)
    regex_template = re.sub(r'\{(.*?)\}', r'(.*?)', pattern)
    # 完整匹配sample_input
    match_result = re.fullmatch(regex_template, sample_input)
    
    if not match_result:
        return "{}"
    
    # 组装成要求的键值对字符串
    key_value_pairs = [f"{key}=\"{value}\"" for key, value in zip(params, match_result.groups())]
    return f'{{{", ".join(key_value_pairs)}}}'

方式二：返回Map字典（更便于后续数据处理）

def extract_params_map(pattern, sample_input):
    params = re.findall(r'\{(.*?)\}', pattern)
    param_dict = {}
    
    if params:
        regex_template = re.sub(r'\{(.*?)\}', r'(.*?)', pattern)
        match_result = re.fullmatch(regex_template, sample_input)
        if match_result:
            param_dict = dict(zip(params, match_result.groups()))
    
    return param_dict

3. 注册UDF并应用到DataFrame #

# 注册字符串格式的UDF
extract_str_udf = udf(extract_params_str, StringType())
# 注册Map格式的UDF
extract_map_udf = udf(extract_params_map, MapType(StringType(), StringType()))

# 生成结果DataFrame
result_df = df.withColumn("expected_result", extract_str_udf("pattern", "sample_input")) \
              .withColumn("param_map", extract_map_udf("pattern", "sample_input"))

# 查看结果
result_df.show(truncate=False)

关键逻辑说明 #

• 用正则\{(.*?)\}提取pattern中的所有参数名，支持每行不同的参数列表
• 将pattern转换为正则模板时，用捕获组(.*?)替代{param}，实现对sample_input对应位置值的捕获
• 通过re.fullmatch确保sample_input完全符合pattern的结构，避免部分匹配导致的错误

注意事项 #

• 如果pattern和sample_input结构不匹配（比如示例中第二行pattern多了&timeout），fullmatch会返回空，此时可以根据需求改为re.search实现部分匹配，或者添加额外的容错逻辑
• 如果参数值包含特殊字符（如.、-），可以调整正则捕获组的规则，比如用([a-zA-Z0-9\-]+)替代(.*?)，避免过度匹配

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