使用Aho-Corasick算法征服字符串搜索

本文概述

Knuth-Morris-Pratt(KMP)算法
Aho-Corasick算法
并行化Aho-Corasick算法
通用的字符串搜索算法

操纵字符串并在其中搜索模式是数据科学中的基本任务, 并且是任何程序员的典型任务。
高效的字符串算法在许多数据科学过程中都起着重要作用。通常, 正是这些因素使此类过程足以实际应用。

文章图片
在本文中, 你将学习用于在大量文本中搜索模式的最强大的算法之一：Aho-Corasick算法。该算法使用trie数据结构(发音为” try” )来跟踪搜索模式, 并使用一种简单的方法来有效地找到在任何文本块中出现的给定模式集的所有出现。
srcmini工程博客上的上一篇文章演示了针对同一问题的字符串搜索算法。本文采用的方法提供了改进的计算复杂性。
Knuth-Morris-Pratt(KMP)算法为了了解如何有效地在文本中查找多个模式, 我们需要首先解决一个更简单的问题：在给定的文本中查找单个模式。
假设我们有一个长度为N的大文本块和一个长度为M的模式(我们要在文本中查找)。无论我们要查找该模式的单个出现还是所有出现, 我们都可以使用KMP算法实现O(N + M)的计算复杂度。
前缀功能
KMP算法通过计算要搜索的模式的前缀函数来工作。前缀函数为模式的每个前缀预先计算后备位置。
让我们将搜索模式定义为一个字符串, 标记为S。对于每个子字符串S [0..i](其中i> = 1), 我们都会找到该字符串的最大前缀, 该前缀也恰好是该字符串的后缀。我们将标记此前缀P [i]的长度。
对于模式” abracadabra” , 前缀函数将产生以下后备位置：

Index (i)	0	1	2	3	4	5	6	7	8	9	10
字符	一个	b	[R	一个	C	一个	d	一个	b	[R	一个
Prefix Length (P[i])	0	0	0	1	0	1	0	1	2	3	4

前缀函数标识模式的有趣特征。
让我们以模式的特定前缀为例：” abracadab” 。此前缀的前缀函数值为2。这表明对于该前缀” abracadab” , 存在一个长度为2的后缀, 该后缀与长度为2的前缀完全匹配(即, 模式以” ab” 开头, 前缀以” ab” 结尾)。此前缀的最长此类匹配项。
实作这是一个C＃函数, 可用于计算任何字符串的前缀函数：

public int[] CalcPrefixFunction(String s) { int[] result = new int[s.Length]; // array with prefix function valuesresult[0] = 0; // the prefix function is always zero for the first symbol (its degenerate case) int k = 0; // current prefix function value for (int i = 1; i < s.Length; i++) { // We're jumping by prefix function values to try smaller prefixes // Jumping stops when we find a match, or reach zero // Case 2 if we stop at a zero-length prefix // Case 3 if we stop at a match while (k > 0 & & s[i] != s[k]) k = result[k - 1]; if (s[k] == s[i]) k++; // we've found the longest prefix - case 1 result[i] = k; // store this result in the array }return result; }

在稍长的模式” abcdabcabcdabcdab” 上运行此函数将产生以下结果：

Index (i)	0	1	2	3	4	5	6	7	8	9	10	11	12	13	14	15	16
字符	一个	b	C	d	一个	b	C	一个	b	C	d	一个	b	C	d	一个	b
Prefix Function (P[i])	0	0	0	0	1	2	3	1	2	3	4	5	6	7	4	5	6

计算复杂度尽管存在两个嵌套循环, 但前缀函数的复杂度仅为O(M), 其中M是模式S的长度。
通过观察循环的工作原理可以很容易地解释这一点。
通过i进行的所有外循环迭代可以分为三种情况：

将k加1。该循环完成一次迭代。
不会更改k的零值。该循环完成一次迭代。
不会更改或减少k的正值。

【使用Aho-Corasick算法征服字符串搜索】前两种情况最多可以运行M次。
对于第三种情况, 我们定义P(s, i)= k1, P(s, i + 1)= k2, k2 < = k1。在每种情况下, 都应先出现第一种情况的k1-k2次。减少的次数不超过k1-k2 +1。总的来说, 我们的迭代次数不超过2 *M。
第二个例子的解释我们来看第二个示例模式” abcdabcabcdabcdab” 。这是前缀函数逐步处理它的方式：

对于空的子字符串和长度为1的子字符串” a” , 前缀函数的值设置为零。 (k = 0)
查看子字符串” ab” 。 k的当前值为零, 字符” b” 不等于字符” a” 。在这里, 我们有上一节中的第二种情况。 k的值保持为零, 子串” ab” 的前缀函数的值也为零。
子字符串” abc” 和” abcd” 的情况相同。没有前缀也是这些子字符串的后缀。它们的值保持为零。
现在让我们看一个有趣的案例, 子字符串” abcda” 。 k的当前值仍为零, 但子字符串的最后一个字符与其第一个字符匹配。这将触发s [k] == s [i]的条件, 其中k == 0且i ==4。数组的索引为零, 而k为最大长度前缀的下一个字符的索引。这意味着我们发现子字符串的最大长度前缀也为后缀。我们有第一种情况, 其中k的新值是1, 因此前缀函数P(” abcda” )的值是1。
接下来的两个子字符串P(” abcdab” )= 2和P(” abcdabc” )= 3也会发生相同的情况。在这里, 我们在文本中搜索模式, 逐字符比较字符串。假设模式的前七个字符与已处理文本的大约七个连续字符匹配, 但是在第八个字符上则不匹配。接下来应该怎么办？对于简单的字符串匹配, 我们应该返回七个字符, 然后从模式的第一个字符开始重新进行比较。使用前缀函数值(此处P(” abcdabc” )= 3), 我们知道我们的三个字符的后缀已经与文本的三个字符匹配。并且, 如果文本中的下一个字符为” d” , 则模式和文本中子字符串的匹配子字符串的长度将增加为四个(” abcd” )。否则, P(” abc” )= 0, 我们将从模式的第一个字符开始比较过程。但是重要的是, 在处理文本期间我们不要退货。
下一个子字符串是” abcdabca” 。在前面的子字符串中, 前缀函数等于3。这意味着k = 3大于零, 同时前缀(s [k] = s [3] =” d” )的下一个字符与后缀()的下一个字符不匹配s [i] = s [7] =” a” )。这意味着我们触发了s [k]！= s [i]的条件, 并且前缀” abcd” 不能作为字符串的后缀。我们应该减小k的值, 并在可能的情况下使用前一个前缀进行比较。如上所述, 数组是零索引的, 而k是我们从前缀检查的下一个字符的索引。当前匹配的前缀的最后一个索引是k-1。对于当前匹配的前缀k = result [k-1], 我们采用前缀函数的值。在我们的情况下(第三种情况), 最大前缀的长度将减少为零, 然后在下一行将增加为一个, 因为” a” 是最大前缀, 也是我们子字符串的后缀。
(在这里, 我们继续计算过程, 直到遇到一个更有趣的情况为止。)
我们开始处理以下子字符串：” abcdabcabcdabcd” 。 k的当前值为7。与上面的” abcdabca” 一样, 我们遇到了不匹配的情况：由于字符” a” (第七个字符)不等于字符” d” , 因此子字符串” abcdabca” 不能作为字符串的后缀。现在, 我们已经获得了前缀函数” abcdabc” (三个)的已计算值, 并且现在有了一个匹配项：前缀” abcd” 也是字符串的后缀。它的最大前缀和子字符串的prefix函数的值为4, 因为这就是k的当前值。
我们继续此过程, 直到模式结束。

简而言之：两个循环最多进行3 M次迭代, 这证明复杂度为O(M)。内存使用量也为O(M)。

KMP算法实现

public int KMP(String text, String s) { int[] p = CalcPrefixFunction(s); // Calculate prefix function for a pattern string // The idea is the same as in the prefix function described above, but now // we're comparing prefixes of text and pattern.// The value of maximum-length prefix of the pattern string that was found // in the text: int maxPrefixLength = 0; for (int i = 0; i < text.Length; i++) { // As in the prefix function, we're jumping by prefixes until k is // greater than 0 or we find a prefix that has matches in the text. while (maxPrefixLength > 0 & & text[i] != s[maxPrefixLength]) maxPrefixLength = p[maxPrefixLength - 1]; // If a match happened, increase the length of the maximum-length // prefix. if (s[maxPrefixLength] == text[i]) maxPrefixLength++; // If the prefix length is the same length as the pattern string, it // means that we have found a matching substring in the text. if (maxPrefixLength == s.Length) { // We can return this value or perform this operation. int idx = i - s.Length + 1; // Get the previous maximum-length prefix and continue search. maxPrefixLength = p[maxPrefixLength - 1]; } } return -1; }

上面的算法逐个字符地遍历文本, 并尝试增加我们的模式和文本中一些连续字符的最大前缀。万一失败, 我们将不会把我们的立场恢复到本文前面的部分。我们知道找到的模式子字符串的最大前缀；该前缀也是此找到的子字符串的后缀, 我们可以简单地继续搜索。
该函数的复杂度与前缀函数的复杂度相同, 从而使总体计算复杂度为O(N + M)和O(M)内存。

琐事：.NET框架中的方法String.IndexOf()和String.Contains()具有一个复杂的算法。

Aho-Corasick算法现在, 我们想对多个模式执行相同的操作。
假设有M个长度为L1, L2, … , Lm的模式。我们需要从字典中找到长度为N的所有模式匹配项。
一个简单的解决方案是从第一部分采用任何算法并运行M次。我们具有O(N + L1 + N + L2 +…+ N + Lm)的复杂度, 即O(M * N + L)。
任何足够严重的测试都会杀死该算法。
以一本包含1, 000个最常见英语单词的字典作为模式, 并使用它来搜索托尔斯泰的《战争与和平》的英文版本, 将花费相当长的时间。这本书长于300万个字符。
如果我们使用10, 000个最常用的英语单词, 该算法的运行速度将降低10倍左右。显然, 对于大于此输入的输入, 执行时间也会增加。
这就是Aho-Corasick算法发挥作用的地方。
Aho-Corasick算法的复杂度为O(N + L + Z), 其中Z是匹配计数。该算法由Alfred V.Aho和Margaret J.Corasick于1975年发明。
实作
在这里, 我们需要一个特里树, 并将与上述前缀函数类似的想法添加到我们的算法中。我们将计算整个字典的前缀函数值。
特里树中的每个顶点将存储以下信息：

public class Vertex { public Vertex() { Children = new Hashtable(); Leaf = false; Parent = -1; SuffixLink = -1; WordID = -1; EndWordLink= -1; }// Links to the child vertexes in the trie: // Key: A single character // Value: The ID of vertex public Hashtable Children; // Flag that some word from the dictionary ends in this vertex public bool Leaf; // Link to the parent vertex public int Parent; // Char which moves us from the parent vertex to the current vertex public char ParentChar; // Suffix link from current vertex (the equivalent of P[i] from the KMP algorithm) public int SuffixLink; // Link to the leaf vertex of the maximum-length word we can make from the current prefix public int EndWordLink; // If the vertex is the leaf, we store the ID of the word public int WordID; }

有多种实现子链接的方法。对于数组, 该算法将具有O(N + L + Z)的复杂度, 但这将具有O(L * q)的额外存储要求, 其中q是字母的长度, 因为一个节点可以拥有的最大子节点数。
如果我们使用某种结构, 并通过O(log(q))访问其元素, 则会有O(L)的额外内存需求, 但是整个算法的复杂度将为O((N + L)* log(q) + Z)。
在哈希表的情况下, 我们有O(L)个额外的内存, 整个算法的复杂度将为O(N + L + Z)。
本教程使用哈希表, 因为它也可以使用不同的字符集(例如汉字)。
我们已经有一个顶点结构。接下来, 我们将定义一个顶点列表并初始化trie的根节点。

public class Aho { List< Vertex> Trie; List< int> WordsLength; int size = 0; int root = 0; public Aho() { Trie = new List< Vertex> (); WordsLength = new List< int> (); Init(); } private void Init() { Trie.Add(new Vertex()) size++; } }

然后, 我们将所有模式添加到特里。为此, 我们需要一种将单词添加到trie的方法：

public void AddString(String s, int wordID) { int curVertex = root; for (int i = 0; i < s.Length; ++i) // Iterating over the string's characters { char c = s[i]; // Checking if a vertex with this edge exists in the trie: if (!Trie[curVertex].Children.ContainsKey(c)) { Trie.Add(new Vertex()); Trie[size].SuffixLink = -1; // If not - add vertex Trie[size].Parent = curVertex; Trie[size].ParentChar = c; Trie[curVertex].Children[c] = size; size++; } curVertex = (int)Trie[curVertex].Children[c]; // Move to the new vertex in the trie } // Mark the end of the word and store its ID Trie[curVertex].Leaf = true; Trie[curVertex].WordID = wordID; WordsLength.Add(s.Length); }

此时, 所有模式字都在数据结构中。这需要O(L)的附加内存。
接下来, 我们需要计算所有后缀链接和字典条目链接。
为了使内容清晰易懂, 我将遍历树的根部到叶子, 并像对KMP算法那样进行类似的计算, 但是与KMP算法相反, 在该算法中, 我们找到了最大长度前缀, 它也是相同子字符串的后缀, 现在我们将找到当前子字符串的最大长度后缀, 它也是trie中某些模式的前缀。该函数的值将不是找到的后缀的长度。它是当前子字符串最大后缀的最后一个字符的链接。这就是我的顶点后缀链接。
我将按级别处理顶点。为此, 我将使用广度优先搜索(BFS)算法：

文章图片
下面是该遍历的实现：

public void PrepareAho() { Queue< int> vertexQueue = new Queue< int> (); vertexQueue.Enqueue(root); while (vertexQueue.Count > 0) { int curVertex = vertexQueue.Dequeue(); CalcSuffLink(curVertex); foreach (char key in Trie[curVertex].Children.Keys) { vertexQueue.Enqueue((int)Trie[curVertex].Children[key]); } } }

下面是CalcSuffLink方法, 用于计算每个顶点的后缀链接(即Trie中每个子字符串的前缀函数值)：

public void CalcSuffLink(int vertex) { // Processing root (empty string) if (vertex == root) { Trie[vertex].SuffixLink = root; Trie[vertex].EndWordLink = root; return; } // Processing children of the root (one character substrings) if (Trie[vertex].Parent == root) { Trie[vertex].SuffixLink = root; if (Trie[vertex].Leaf) Trie[vertex].EndWordLink = vertex; else Trie[vertex].EndWordLink = Trie[Trie[vertex].SuffixLink].EndWordLink; return; } // Cases above are degenerate cases as for prefix function calculation; the // value is always 0 and links to the root vertex. // To calculate the suffix link for the current vertex, we need the suffix // link for the parent of the vertex and the character that moved us to the // current vertex. int curBetterVertex = Trie[Trie[vertex].Parent].SuffixLink; char chVertex = Trie[vertex].ParentChar; // From this vertex and its substring we will start to look for the maximum // prefix for the current vertex and its substring. while (true) { // If there is an edge with the needed char, we update our suffix link // and leave the cycle if (Trie[curBetterVertex].Children.ContainsKey(chVertex)) { Trie[vertex].SuffixLink = (int)Trie[curBetterVertex].Children[chVertex]; break; } // Otherwise, we are jumping by suffix links until we reach the root // (equivalent of k == 0 in prefix function calculation) or we find a // better prefix for the current substring. if (curBetterVertex == root) { Trie[vertex].SuffixLink = root; break; } curBetterVertex = Trie[curBetterVertex].SuffixLink; // Go back by sufflink } // When we complete the calculation of the suffix link for the current // vertex, we should update the link to the end of the maximum length word // that can be produced from the current substring. if (Trie[vertex].Leaf) Trie[vertex].EndWordLink = vertex; else Trie[vertex].EndWordLink = Trie[Trie[vertex].SuffixLink].EndWordLink; }

此方法的复杂度为O(L)；根据子集合的实现, 复杂度可能为O(L * log(q))。
复杂度证明类似于KMP算法中的复杂度前缀函数证明。
我们来看下图。这是字典{abba, cab, baba, caab, ac, abac, bac}的trie的可视化及其所有计算的信息：

文章图片
特里边缘为深蓝色, 后缀链接为浅蓝色, 字典后缀链接为绿色。与词典条目对应的节点以蓝色突出显示。
现在, 我们只需要一种方法-处理要搜索的文本块：

public int ProcessString(String text) { // Current state value int currentState = root; // Targeted result value int result = 0; for (int j = 0; j < text.Length; j++) { // Calculating new state in the trie while (true) { // If we have the edge, then use it if (Trie[currentState].Children.ContainsKey(text[j])) { currentState = (int)Trie[currentState].Children[text[j]]; break; } // Otherwise, jump by suffix links and try to find the edge with // this char// If there aren't any possible edges we will eventually ascend to // the root, and at this point we stop checking. if (currentState == root) break; currentState = Trie[currentState].SuffixLink; } int checkState = currentState; // Trying to find all possible words from this prefix while (true) { // Checking all words that we can get from the current prefix checkState = Trie[checkState].EndWordLink; // If we are in the root vertex, there are no more matches if (checkState == root) break; // If the algorithm arrived at this row, it means we have found a // pattern match. And we can make additional calculations like find // the index of the found match in the text. Or check that the found // pattern is a separate word and not just, e.g., a substring. result++; int indexOfMatch = j + 1 - WordsLength[Trie[checkState].WordID]; // Trying to find all matched patterns of smaller length checkState = Trie[checkState].SuffixLink; } } return result; }

而且, 现在可以使用了：
在输入时, 我们有一个模式列表：

List< string> patterns;

并搜索文字：

string text;

以下是将其粘合在一起的方法：

// Init the trie structure. As an optional parameter we can put the approximate // size of the trie to allocate memory just once for all nodes. Aho ahoAlg = new Aho(); for (int i = 0; i < patterns.Count; i++) { ahoAlg.AddString(patterns[i], i); // Add all patterns to the structure } // Prepare algorithm for work (calculates all links in the structure): ahoAlg.PrepareAho(); // Process the text. Output might be different; in my case, it's a count of // matches. We could instead return a structure with more detailed information. int countOfMatches = ahoAlg.ProcessString(text);

就是这样！你现在知道了这种简单而强大的算法是如何工作的！
Aho-Corasick非常灵活。搜索模式不必只是单词, 而是我们可以使用整个句子或随机的字符链。
性能
该算法已在Intel Core i7-4702MQ上进行了测试。
为了测试, 我选了两个词典：1, 000个最常用的英语单词和10, 000个最常用的英语单词。
为了将所有这些单词添加到字典中并准备数据结构以与每个字典配合使用, 该算法分别需要55ms和135ms。
该算法在1.0-1.3秒内处理了3-4百万个字符的真实书籍, 而一本包含3000万个字符的书籍则花费了9.6秒。
并行化Aho-Corasick算法与Aho-Corasick算法并行完全不是问题：

文章图片
大文本可以分为多个块, 并且可以使用多个线程来处理每个块。每个线程都可以访问基于字典的生成的trie。
单词在大块之间的边界处分裂怎么办？这个问题很容易解决。
令N为大文本的长度, S为大块的大小, L为字典中最大模式的长度。
现在我们可以使用一个简单的技巧。我们将块尾部分重叠, 例如, 取[S *(i-1), S * i + L-1], 其中i是块的索引。当我们得到一个模式匹配时, 我们可以很容易地获得当前匹配的开始索引, 只需检查该索引是否在块范围内即可, 没有重叠[S *(i-1), S * i-1]。
通用的字符串搜索算法 Aho-Corasick算法是一种功能强大的字符串匹配算法, 可为任何输入提供最佳的复杂性, 并且不需要太多额外的内存。
该算法通常在各种系统中使用, 例如拼写检查器, 垃圾邮件过滤器, 搜索引擎, 生物信息学/ DNA序列搜索等。实际上, 你每天可能会使用的一些流行工具在后台使用了该算法。
KMP算法本身的前缀函数是一个有趣的工具, 它使单模式匹配的复杂度降低到了线性时间。 Aho-Corasick算法采用类似的方法, 并使用trie数据结构对多个模式执行相同的操作。
希望你发现本教程对Aho-Corasick算法有用。