加速 Python 3 中数百万次正则表达式替换
- 2024-12-11 08:47:00
- admin 原创
- 161
问题描述：
我有两个清单：
约 750K 个“句子”（长字符串）的列表
我想从 750K 个句子中删除约 20K 个“单词”的列表
因此，我必须循环遍历 750K个句子并执行大约 20K 次替换，但前提是我的单词实际上是“单词”并且不是较大字符串的一部分。
我通过预编译我的单词来实现这一点，以便它们被单词边界元字符包围：
compiled_words = [re.compile(r'' + word + r'') for word in my20000words]
然后我循环遍历我的“句子”：
import re
for sentence in sentences:
for word in compiled_words:
sentence = re.sub(word, "", sentence)
# put sentence into a growing list
这个嵌套循环每秒处理大约50 个句子，这很好，但处理我所有的句子仍然需要几个小时。
有没有办法使用该
str.replace方法（我认为速度更快），但仍然要求替换仅发生在单词边界？或者，有没有办法加快这个
re.sub方法？我已经通过跳过re.sub单词长度大于句子长度的情况稍微提高了速度，但改进不大。
我正在使用 Python 3.5.2
解决方案 1：
结论
如果您想要最快的基于正则表达式的解决方案，请使用此方法。对于与 OP 类似的数据集，它比接受的答案快大约 1000 倍。
如果您不关心正则表达式，请使用这个基于集合的版本，它比正则表达式联合快 2000 倍。
使用 Trie 优化正则表达式
简单的正则表达式联合方法在处理许多禁用词时会变得很慢，因为正则表达式引擎在优化模式方面做得不是很好。
可以创建一个包含所有禁用词的Trie并编写相应的正则表达式。生成的 Trie 或正则表达式实际上不是人类可读的，但它们确实允许非常快速的查找和匹配。
例子
['foobar', 'foobah', 'fooxar', 'foozap', 'fooza']
该列表被转换为 trie：
{
'f': {
'o': {
'o': {
'x': {
'a': {
'r': {
'': 1
}
}
},
'b': {
'a': {
'r': {
'': 1
},
'h': {
'': 1
}
}
},
'z': {
'a': {
'': 1,
'p': {
'': 1
}
}
}
}
}
}
}
然后到这个正则表达式模式：
r"foo(?:ba[hr]|xar|zap?)"
巨大的优势在于，为了测试是否zoo匹配，正则表达式引擎只需要比较第一个字符（不匹配），而不是尝试 5 个单词。对于 5 个单词来说，这是一个过度的预处理，但对于数千个单词来说，它显示出令人鼓舞的结果。
请注意，使用(?:)非捕获组是因为：
foobar|baz匹配foobar或baz，但不会匹配foobazfoo(bar|baz)会将不需要的信息保存到捕获组。
代码
这是一个稍微修改过的要点，我们可以将其用作trie.py库：
import re
class Trie():
"""Regex::Trie in Python. Creates a Trie out of a list of words. The trie can be exported to a Regex pattern.
The corresponding Regex should match much faster than a simple Regex union."""
def __init__(self):
self.data = {}
def add(self, word):
ref = self.data
for char in word:
ref[char] = char in ref and ref[char] or {}
ref = ref[char]
ref[''] = 1
def dump(self):
return self.data
def quote(self, char):
return re.escape(char)
def _pattern(self, pData):
data = pData
if "" in data and len(data.keys()) == 1:
return None
alt = []
cc = []
q = 0
for char in sorted(data.keys()):
if isinstance(data[char], dict):
try:
recurse = self._pattern(data[char])
alt.append(self.quote(char) + recurse)
except:
cc.append(self.quote(char))
else:
q = 1
cconly = not len(alt) > 0
if len(cc) > 0:
if len(cc) == 1:
alt.append(cc[0])
else:
alt.append('[' + ''.join(cc) + ']')
if len(alt) == 1:
result = alt[0]
else:
result = "(?:" + "|".join(alt) + ")"
if q:
if cconly:
result += "?"
else:
result = "(?:%s)?" % result
return result
def pattern(self):
return self._pattern(self.dump())
测试
这是一个小测试（与这个相同）：
# Encoding: utf-8
import re
import timeit
import random
from trie import Trie
with open('/usr/share/dict/american-english') as wordbook:
banned_words = [word.strip().lower() for word in wordbook]
random.shuffle(banned_words)
test_words = [
("Surely not a word", "#surely_NöTäWORD_so_regex_engine_can_return_fast"),
("First word", banned_words[0]),
("Last word", banned_words[-1]),
("Almost a word", "couldbeaword")
]
def trie_regex_from_words(words):
trie = Trie()
for word in words:
trie.add(word)
return re.compile(r"" + trie.pattern() + r"", re.IGNORECASE)
def find(word):
def fun():
return union.match(word)
return fun
for exp in range(1, 6):
print("
TrieRegex of %d words" % 10**exp)
union = trie_regex_from_words(banned_words[:10**exp])
for description, test_word in test_words:
time = timeit.timeit(find(test_word), number=1000) * 1000
print(" %s : %.1fms" % (description, time))
它输出：
TrieRegex of 10 words
Surely not a word : 0.3ms
First word : 0.4ms
Last word : 0.5ms
Almost a word : 0.5ms
TrieRegex of 100 words
Surely not a word : 0.3ms
First word : 0.5ms
Last word : 0.9ms
Almost a word : 0.6ms
TrieRegex of 1000 words
Surely not a word : 0.3ms
First word : 0.7ms
Last word : 0.9ms
Almost a word : 1.1ms
TrieRegex of 10000 words
Surely not a word : 0.1ms
First word : 1.0ms
Last word : 1.2ms
Almost a word : 1.2ms
TrieRegex of 100000 words
Surely not a word : 0.3ms
First word : 1.2ms
Last word : 0.9ms
Almost a word : 1.6ms
供参考，正则表达式的开头如下：
（？：a（？：（？：\'s|a（？：\'s|chen|liyah（？：\'s）？|r（？：dvark（？：（？：\'s|s））？|on））|b（？：\'s|a（？：c（？：us（？：（？：\'s|es））？|[ik]）|ft|lone（？：（？：\'s|s））？|ndon（？：（？：ed|ing|ment（？：\'s）？|s））？|s（？：e（？：（？：ment（？：\'s）？|[ds]））？|h（？：（？：e[ds]|ing））？|ing）|t（？：e（？：（？：ment（？：\'s）？|[d s]))?|ing|toir(?:(?:\'s|s))?))|b(?:as(?:id)?|e(?:ss(?:(?:\'s|es))?|y(?:(?:\'s|s))?)|ot(?:(?:\'s|t(?:\'s)?|s))?|reviat(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?))|y(?:\'s)?|\é(?:(?:\'s|s))?)|d(?:icat(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?))|om(?:en(?:(?:\'s|s))?|ina l)|u(?:ct(?:(?:ed|i(?:ng|on(?:(?:\'s|s))?)|or(?:(?:\'s|s))?|s))?|l(?:\'s)?))|e(?:(?:\'s|am|l(?:(?:\'s|ard|son(?:\'s)?))?|r(?:deen(?:\'s)?|nathy(?:\'s)?|ra(?:nt|tion(?:(?:\'s|s))?))|t(?:(?:t(?:e(?:r(?:(?:\'s|s))?|d)|ing|or(?:(?:\'s|s))?)|s))?|yance(?: \'s)?|d))?|hor(?:(?:r(?:e(?:n(?:ce(?:\'s)?|t)|d)|ing)|s))?|i(?:d(?:e[ds]?|ing|jan(?:\'s)?)|gail|l(?:ene|it(?:ies|y(?:\'s)?)))|j(?:ect(?:ly)?|ur(?:ation(?:(?:\'s|s))?|e[ds]?|ing))|l(?:a(?:tive(?:(?:\'s|s))?|ze)|e(?:(?:st|r))?|oom|ution(?:(?:\'s|s))?|y) |m\'s|n(?:e(?:gat(?:e[ds]?|i(?:ng|on(?:\'s)?))|r(?:\'s)?)|ormal(?:(?:it(?:ies|y(?:\'s)?)|ly))?)|o(?:ard|de(?:(?:\'s|s))?|li(?:sh(?:(?:e[ds]|ing))?|tion(?:(?:\'s|ist(?:(?:\'s|s))?))?)|mina(?:bl[ey]|t(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?)))|r(?:igin(?:al(? :(?:\'s|s))?|e(?:(?:\'s|s))?)|t(?:(?:ed|i(?:ng|on(?:(?:\'s|ist(?:(?:\'s|s))?|s))?|ve)|s))?)|u(?:nd(?:(?:ed|ing|s))?|t)|ve(?:(?:\'s|board))?)|r(?:a(?:cadabra(?:\'s)?|d(?:e[ds]?|ing)|ham(?:\'s)?|m(?:(?:\'s|s))?|si(?:on(?:(?:\'s|s))?|ve(?:(?:\'s|ly|ness( ?:\'s)?|s))?))|east|idg(?:e(?:(?:ment(?:(?:\'s|s))?|[ds]))?|ing|ment(?:(?:\'s|s))?)|o(?:ad|gat(?:e[ds]?|i(?:ng|on(?:(?:\'s|s))?)))|upt(?:(?:e(?:st|r)|ly|ness(?:\'s)?))?)|s(?:alom|c(?:ess(?:(?:\'s|e[ds]|ing))?|issa(?:(?:\'s|[es]))?|ond(?:(?:ed|ing|s)) )|en(?:ce(?:(?:\'s|s))?|t(?:(?:e(?:e(?:(?:\'s|ism(?:\'s)?|s))?|d)|ing|ly|s))?)|inth(?:(?:\'s|e(?:\'s)?))?|o(?:l(?:ut(?:e(?:(?:\'s|ly|st?))?|i(?:on(?:\'s)?|sm(?:\'s)?))|v(?:e[ds]?|ing))|r(?:b(?:(?:e(?:n(?:cy(?:\'s)?|t(?:(?:\'s|s))?)|d)|ing|s))?|pti...
这确实难以阅读，但是对于 100000 个禁用词的列表，这个 Trie 正则表达式比简单的正则表达式联合快 1000 倍！
下面是使用trie-python-graphviz和 graphviz导出的完整 trie 图表twopi：
解决方案 2：
结论
如果您想要最快的解决方案，请使用此方法（使用集合查找）。对于与 OP 类似的数据集，它比接受的答案快大约 2000 倍。
如果您坚持使用正则表达式进行查找，请使用这个基于 trie 的版本，它仍然比正则表达式联合快 1000 倍。
理论
如果您的句子不是巨大的字符串，那么每秒处理超过 50 个句子可能是可行的。
如果将所有禁用词保存到一个集合中，那么检查该集合中是否包含另一个词就会非常快。
将逻辑打包成一个函数，将该函数作为参数传入re.sub，就完成了！
代码
import re
with open('/usr/share/dict/american-english') as wordbook:
banned_words = set(word.strip().lower() for word in wordbook)
def delete_banned_words(matchobj):
word = matchobj.group(0)
if word.lower() in banned_words:
return ""
else:
return word
sentences = ["I'm eric. Welcome here!", "Another boring sentence.",
"GiraffeElephantBoat", "sfgsdg sdwerha aswertwe"] * 250000
word_pattern = re.compile('w+')
for sentence in sentences:
sentence = word_pattern.sub(delete_banned_words, sentence)
转换后的句子为：
' . !
.
GiraffeElephantBoat
sfgsdg sdwerha aswertwe
注意：
搜索不区分大小写（感谢
lower()）用替换一个单词
""可能会留下两个空格（如您的代码所示）使用 python3，
w+还匹配重音字符（例如"ångström"）。任何非单词字符（制表符、空格、换行符、标记等）都将保持不变。
表现
有百万个句子，banned_words近十万个单词，脚本运行时间不到 7 秒。
相比之下，Liteye 的答案需要 160 秒才能读完 1 万个句子。
加上n字数和m禁言量，OP和Liteye的代码就是O(n*m)。
相比之下，我的代码应该在 中运行O(n+m)。考虑到句子数量远多于禁用词数量，算法变为O(n)。
正则表达式联合测试
使用模式进行正则表达式搜索的复杂度是多少'(word1|word2|...|wordN)'？是O(N)还是O(1)？
正则表达式引擎的工作方式很难掌握，所以让我们编写一个简单的测试。
此代码将10**i随机英文单词提取到列表中。它创建相应的正则表达式联合，并使用不同的单词对其进行测试：
one 显然不是一个单词（它以 开头
#）one 是列表中的第一个单词
one 是列表中的最后一个单词
看起来像一个词但实际上不是
import re
import timeit
import random
with open('/usr/share/dict/american-english') as wordbook:
english_words = [word.strip().lower() for word in wordbook]
random.shuffle(english_words)
print("First 10 words :")
print(english_words[:10])
test_words = [
("Surely not a word", "#surely_NöTäWORD_so_regex_engine_can_return_fast"),
("First word", english_words[0]),
("Last word", english_words[-1]),
("Almost a word", "couldbeaword")
]
def find(word):
def fun():
return union.match(word)
return fun
for exp in range(1, 6):
print("
Union of %d words" % 10**exp)
union = re.compile(r"(%s)" % '|'.join(english_words[:10**exp]))
for description, test_word in test_words:
time = timeit.timeit(find(test_word), number=1000) * 1000
print(" %-17s : %.1fms" % (description, time))
它输出：
First 10 words :
["geritol's", "sunstroke's", 'fib', 'fergus', 'charms', 'canning', 'supervisor', 'fallaciously', "heritage's", 'pastime']
Union of 10 words
Surely not a word : 0.7ms
First word : 0.8ms
Last word : 0.7ms
Almost a word : 0.7ms
Union of 100 words
Surely not a word : 0.7ms
First word : 1.1ms
Last word : 1.2ms
Almost a word : 1.2ms
Union of 1000 words
Surely not a word : 0.7ms
First word : 0.8ms
Last word : 9.6ms
Almost a word : 10.1ms
Union of 10000 words
Surely not a word : 1.4ms
First word : 1.8ms
Last word : 96.3ms
Almost a word : 116.6ms
Union of 100000 words
Surely not a word : 0.7ms
First word : 0.8ms
Last word : 1227.1ms
Almost a word : 1404.1ms
因此看起来搜索具有某种'(word1|word2|...|wordN)'模式的单个单词有：
O(1)最好的情况O(n/2)平均情况，仍然是O(n)O(n)最坏的情况
这些结果与简单的循环搜索一致。
正则表达式联合的一个更快捷的替代方法是从 trie创建正则表达式模式。
解决方案 3：
您可以尝试的一件事是编译一个类似这样的单一模式"(word1|word2|word3)"。
因为re依赖 C 代码进行实际匹配，所以节省的成本非常可观。
正如@pvg 在评论中指出的那样，它也受益于单次匹配。
如果你的词不是正则表达式，Eric 的回答会更快。
解决方案 4：
您可能想要尝试的一件事是预处理句子以对单词边界进行编码。基本上，通过按单词边界进行拆分，将每个句子变成单词列表。
这应该更快，因为要处理一个句子，你只需要逐步检查每个单词并检查它是否匹配。
目前，正则表达式搜索每次都必须再次遍历整个字符串，寻找单词边界，然后在下一次遍历之前“丢弃”本次工作的结果。
解决方案 5：
好吧，这是一个快速简便的解决方案，带有测试集。
最佳策略：
re.sub("w+",repl,sentence)搜索单词。“repl” 可以是可调用函数。我使用了一个执行字典查找的函数，该字典包含要搜索和替换的单词。
这是最简单、最快的解决方案（参见
replace4下面示例代码中的函数）。
次优策略：
其思路是使用 把句子拆分成单词，
re.split同时保留分隔符以便稍后重建句子。然后，使用简单的字典查找完成替换。实现：（参见
replace3下面示例代码中的函数）。
示例功能的时间安排：
replace1: 0.62 sentences/s
replace2: 7.43 sentences/s
replace3: 48498.03 sentences/s
replace4: 61374.97 sentences/s (...and 240,000/s with PyPy)
...和代码：
#! /bin/env python3
# -*- coding: utf-8
import time, random, re
def replace1( sentences ):
for n, sentence in enumerate( sentences ):
for search, repl in patterns:
sentence = re.sub( "\\b"+search+"\\b", repl, sentence )
def replace2( sentences ):
for n, sentence in enumerate( sentences ):
for search, repl in patterns_comp:
sentence = re.sub( search, repl, sentence )
def replace3( sentences ):
pd = patterns_dict.get
for n, sentence in enumerate( sentences ):
#~ print( n, sentence )
# Split the sentence on non-word characters.
# Note: () in split patterns ensure the non-word characters ARE kept
# and returned in the result list, so we don't mangle the sentence.
# If ALL separators are spaces, use string.split instead or something.
# Example:
#~ >>> re.split(r"([^w]+)", "ab céé? . d2eéf")
#~ ['ab', ' ', 'céé', '? . ', 'd2eéf']
words = re.split(r"([^w]+)", sentence)
# and... done.
sentence = "".join( pd(w,w) for w in words )
#~ print( n, sentence )
def replace4( sentences ):
pd = patterns_dict.get
def repl(m):
w = m.group()
return pd(w,w)
for n, sentence in enumerate( sentences ):
sentence = re.sub(r"w+", repl, sentence)
# Build test set
test_words = [ ("word%d" % _) for _ in range(50000) ]
test_sentences = [ " ".join( random.sample( test_words, 10 )) for _ in range(1000) ]
# Create search and replace patterns
patterns = [ (("word%d" % _), ("repl%d" % _)) for _ in range(20000) ]
patterns_dict = dict( patterns )
patterns_comp = [ (re.compile("\\b"+search+"\\b"), repl) for search, repl in patterns ]
def test( func, num ):
t = time.time()
func( test_sentences[:num] )
print( "%30s: %.02f sentences/s" % (func.__name__, num/(time.time()-t)))
print( "Sentences", len(test_sentences) )
print( "Words ", len(test_words) )
test( replace1, 1 )
test( replace2, 10 )
test( replace3, 1000 )
test( replace4, 1000 )
编辑：您还可以在检查是否传递了句子的小写列表并编辑 repl 时忽略小写
def replace4( sentences ):
pd = patterns_dict.get
def repl(m):
w = m.group()
return pd(w.lower(),w)
解决方案 6：
也许 Python 不是合适的工具。下面是使用 Unix 工具链的示例
sed G file |
tr ' ' '
' |
grep -vf blacklist |
awk -v RS= -v OFS=' ' '{$1=$1}1'
假设您的黑名单文件已预处理并添加了单词边界。步骤如下：将文件转换为双倍行距，将每个句子拆分为每行一个单词，从文件中批量删除黑名单单词，然后合并回行。
这应该至少运行得更快一个数量级。
用于预处理单词黑名单文件（每行一个单词）
sed 's/.*/\\b&\\b/' words > blacklist
解决方案 7：
这个怎么样：
#!/usr/bin/env python3
from __future__ import unicode_literals, print_function
import re
import time
import io
def replace_sentences_1(sentences, banned_words):
# faster on CPython, but does not use as the word separator
# so result is slightly different than replace_sentences_2()
def filter_sentence(sentence):
words = WORD_SPLITTER.split(sentence)
words_iter = iter(words)
for word in words_iter:
norm_word = word.lower()
if norm_word not in banned_words:
yield word
yield next(words_iter) # yield the word separator
WORD_SPLITTER = re.compile(r'(W+)')
banned_words = set(banned_words)
for sentence in sentences:
yield ''.join(filter_sentence(sentence))
def replace_sentences_2(sentences, banned_words):
# slower on CPython, uses as separator
def filter_sentence(sentence):
boundaries = WORD_BOUNDARY.finditer(sentence)
current_boundary = 0
while True:
last_word_boundary, current_boundary = current_boundary, next(boundaries).start()
yield sentence[last_word_boundary:current_boundary] # yield the separators
last_word_boundary, current_boundary = current_boundary, next(boundaries).start()
word = sentence[last_word_boundary:current_boundary]
norm_word = word.lower()
if norm_word not in banned_words:
yield word
WORD_BOUNDARY = re.compile(r'')
banned_words = set(banned_words)
for sentence in sentences:
yield ''.join(filter_sentence(sentence))
corpus = io.open('corpus2.txt').read()
banned_words = [l.lower() for l in open('banned_words.txt').read().splitlines()]
sentences = corpus.split('. ')
output = io.open('output.txt', 'wb')
print('number of sentences:', len(sentences))
start = time.time()
for sentence in replace_sentences_1(sentences, banned_words):
output.write(sentence.encode('utf-8'))
output.write(b' .')
print('time:', time.time() - start)
这些解决方案按单词边界进行拆分，并查找集合中的每个单词。它们应该比单词替代的 re.sub（Liteyes 的解决方案）更快，因为这些解决方案中O(n)n 是由于amortized O(1)集合查找而导致的输入大小，而使用正则表达式替代会导致正则表达式引擎必须检查每个字符上的单词匹配，而不仅仅是单词边界。我的解决方案特别注意保留原始文本中使用的空格（即它不压缩空格并保留制表符、换行符和其他空格字符），但如果您决定不关心它，那么将它们从输出中删除应该相当简单。
我在 corpus.txt 上进行了测试，它是从 Gutenberg 项目下载的多本电子书的串联，banned_words.txt 是从 Ubuntu 的单词表 (/usr/share/dict/american-english) 中随机挑选的 20000 个单词。处理 862462 个句子（其中一半在 PyPy 上）大约需要 30 秒。我将句子定义为用“。”分隔的任何内容。
$ # replace_sentences_1()
$ python3 filter_words.py
number of sentences: 862462
time: 24.46173644065857
$ pypy filter_words.py
number of sentences: 862462
time: 15.9370770454
$ # replace_sentences_2()
$ python3 filter_words.py
number of sentences: 862462
time: 40.2742919921875
$ pypy filter_words.py
number of sentences: 862462
time: 13.1190629005
PyPy 尤其从第二种方法中受益更多，而 CPython 在第一种方法上表现更好。上述代码应该适用于 Python 2 和 3。
解决方案 8：
实用方法
下面介绍的解决方案使用大量内存来存储同一字符串中的所有文本并降低复杂度。如果 RAM 是一个问题，请在使用它之前三思。
使用join/split技巧，您可以完全避免循环，从而加快算法速度。
将句子与句子中不包含的特殊分隔符连接起来：
merged_sentences = ' * '.join(sentences)
|使用“或”正则表达式语句为需要从句子中删除的所有单词编译一个正则表达式：
regex = re.compile(r'({})'.format('|'.join(words)), re.I) # re.I is a case insensitive flag
使用编译后的正则表达式对单词进行下标，并用特殊分隔符将其拆分回分隔的句子：
clean_sentences = re.sub(regex, "", merged_sentences).split(' * ')
表现
"".join复杂度为 O(n)。这很直观，但无论如何，这里有一个简短的引文：
for (i = 0; i < seqlen; i++) {
[...]
sz += PyUnicode_GET_LENGTH(item);
因此，join/split你有 O(words) + 2O(sentences)，这仍然是线性复杂度，而初始方法的复杂度为 2O(N 2 )。
顺便说一句，不要使用多线程。GIL 将阻止每个操作，因为您的任务严格受 CPU 限制，因此 GIL 没有机会被释放，但每个线程都会同时发送滴答声，这会导致额外的努力，甚至导致操作无限。
解决方案 9：
将所有句子连接成一个文档。使用 Aho-Corasick 算法的任何实现（这里有一个）来定位所有“坏”词。遍历文件，替换每个坏词，更新后面找到的单词的偏移量等。
解决方案 10：
使用 LRU 缓存！我在生产环境中将正则表达式搜索速度提高了 60 倍。缓存（又称记忆化）的工作原理是保存昂贵的计算结果，因此如果您的数据集中有重复项（很可能存在），您将获得最佳速度。
import re
from functools import lru_cache
@lru_cache
def expensive_regex(str):
return re.findall('pattern', str)
查看：仅用两行代码即可将正则表达式速度提高 127 倍，由我本人编写：)
