Python中使用re.compile高效处理万级字符串的优化方案咨询

1. 基础性能优化（最直接有效） #

原代码每次循环都调用re.compile是核心性能浪费——正则编译属于耗时操作，只需执行一次即可。同时你仅需判断是否存在匹配，用re.search()代替findall()能更快终止匹配（找到第一个匹配就停止，无需遍历全文收集所有结果）。

优化后代码：

import re

# 预编译正则表达式，仅执行一次
pattern = re.compile(r"I like to eat (.*?)")
list_of_strings = ["I like to eat meat", "I don't like to eat meat", "I like to eat fish", "I don't like to eat fish"]

# 用search判断是否存在匹配，效率远高于findall
outcome = [x for x in list_of_strings if pattern.search(x) is not None]

2. 多进程并行处理 #

正则匹配属于CPU密集型任务，利用多进程可充分发挥多核CPU优势，适合处理1万条级别的数据。

示例代码：

import re
from concurrent.futures import ProcessPoolExecutor

# 预编译正则
pattern = re.compile(r"I like to eat (.*?)")
list_of_strings = ["I like to eat meat", "I don't like to eat meat", "I like to eat fish", "I don't like to eat fish"]

# 定义单个字符串的匹配检查函数
def check_match(s):
    return pattern.search(s) is not None

# 启动进程池并行处理
with ProcessPoolExecutor() as executor:
    # 批量提交任务，返回匹配结果
    match_results = executor.map(check_match, list_of_strings)

# 过滤出匹配的字符串
outcome = [s for s, is_match in zip(list_of_strings, match_results) if is_match]

3. PySpark分布式处理（适合超大规模数据） #

如果未来数据量远超内存（比如百万级以上），可以用PySpark进行分布式处理，利用集群资源加速。

示例代码：

from pyspark.sql import SparkSession
import re
from pyspark.sql.functions import udf, col
from pyspark.sql.types import BooleanType

# 初始化Spark会话
spark = SparkSession.builder.appName("RegexMatch").getOrCreate()

# 将字符串列表转为Spark DataFrame
list_of_strings = ["I like to eat meat", "I don't like to eat meat", "I like to eat fish", "I don't like to eat fish"]
df = spark.createDataFrame([(text,) for text in list_of_strings], ["content"])

# 预编译正则，定义UDF用于匹配检查
pattern = re.compile(r"I like to eat (.*?)")
@udf(returnType=BooleanType())
def has_matching_pattern(text):
    return pattern.search(text) is not None

# 过滤出匹配的行并提取结果
result_df = df.filter(has_matching_pattern(col("content")))
outcome = [row.content for row in result_df.collect()]

# 关闭Spark会话
spark.stop()

4. PyTorch多进程（仅结合深度学习场景时考虑） #

PyTorch主要用于深度学习任务，若你的正则匹配是DL pipeline的一部分，可以用其多进程工具并行处理，否则没必要单独用它做正则匹配。

示例代码：

import re
import torch
from torch.multiprocessing import Pool

pattern = re.compile(r"I like to eat (.*?)")
list_of_strings = ["I like to eat meat", "I don't like to eat meat", "I like to eat fish", "I don't like to eat fish"]

def check_match(s):
    return pattern.search(s) is not None

# 根据CPU核心数或GPU数量设置进程数
process_num = torch.cuda.device_count() if torch.cuda.is_available() else 4
with Pool(processes=process_num) as pool:
    match_results = pool.map(check_match, list_of_strings)

outcome = [s for s, is_match in zip(list_of_strings, match_results) if is_match]

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