算法数据结构

稀疏相似度：倒排索引

字数统计: 2.9k字   |   阅读时长: 12分
, ,

摘要: 倒排索引简介及其在稀疏相似度中的应用

【对算法，数学，计算机感兴趣的同学，欢迎关注我哈，阅读更多原创文章】
我的网站：潮汐朝夕的生活实验室
我的公众号：算法题刷刷
我的知乎：潮汐朝夕
我的github：FennelDumplings
我的leetcode：FennelDumplings

单词-文档矩阵

单词—文档矩阵是表达两者之间所具有的一种包含关系的概念模型，每列代表一个文档，每行代表一个单词。下面是一个例子：

文档1 文档2 文档3 文档4 文档5
单词1 $\checkmark$ $\checkmark$
单词2 $\checkmark$ $\checkmark$
单词3 $\checkmark$
单词4 $\checkmark$ $\checkmark$
单词5 $\checkmark$
单词6 $\checkmark$

从纵向即文档这个维度来看，每列代表文档包含了哪些单词，比如文档1包含了词汇1和词汇4，而不包含其他单词。从横向即单词这个维度来看，每行代表了哪些文档包含了某个单词。比如对于词汇1来说，文档1和文档4中出现过词汇1，而其他文档不包含词汇1。

搜索引擎的索引其实就是实现单词—文档矩阵的具体数据结构。可以有不同的方式来实现上述概念模型，比如倒排索引、签名文件、后缀树等方式。

本文我们首先介绍倒排索引的相关概念，然后解决力扣面试题 17.26。

倒排索引基本概念

下面我们介绍倒排索引相关的一些概念，概念之间关系如下图：

文档

一般搜索引擎的处理对象是互联网网页，而文档这个概念要更宽泛些，代表以文本形式存在的存储对象。相比网页来说，涵盖更多形式，比如Word、PDF、html、XML等不同格式的文件都可以称为文档，再比如一封邮件、一条短信、一条微博也可以称为文档。

文档集合

由若干文档构成的集合称为文档集合。比如海量的互联网网页或者说大量的电子邮件，都是文档集合的具体例子。

文档编号

在搜索引擎内部，会为文档集合内每个文档赋予一个唯一的内部编号，以此编号来作为这个文档的唯一标识，这样方便内部处理。每个文档的内部编号即称为文档编号，后文有时会用DocID来便捷地代表文档编号。

单词编号

与文档编号类似，搜索引擎内部以唯一的编号来表征某个单词，单词编号可以作为某个单词的唯一表征。

倒排索引

倒排索引是实现单词—文档矩阵的一种具体存储形式。通过倒排索引，可以根据单词快速获取包含这个单词的文档列表。倒排索引主要由两个部分组成：单词词典和倒排文件。

单词词典

搜索引擎通常的索引单位是单词，单词词典是由文档集合中出现过的所有单词构成的字符串集合，单词词典内每条索引项记载单词本身的一些信息及指向倒排列表的指针。

倒排列表

倒排列表记载了出现过某个单词的所有文档的文档列表及单词在该文档中出现的位置信息，每条记录称为一个倒排项（Posting）。根据倒排列表，即可获知哪些文档包含某个单词。

倒排文件

所有单词的倒排列表往往顺序地存储在磁盘的某个文件里，这个文件即被称为倒排文件，倒排文件是存储倒排索引的物理文件。

题目

两个(具有不同单词的)文档的交集(intersection)中元素的个数除以并集(union)中元素的个数，就是这两个文档的相似度。例如，{1, 5, 3} 和 {1, 7, 2, 3} 的相似度是 0.4，其中，交集的元素有 2 个，并集的元素有 5 个。给定一系列的长篇文档，每个文档元素各不相同，并与一个 ID 相关联。它们的相似度非常“稀疏”，也就是说任选 2 个文档，相似度都很接近 0。请设计一个算法返回每对文档的 ID 及其相似度。只需输出相似度大于 0 的组合。请忽略空文档。为简单起见，可以假定每个文档由一个含有不同整数的数组表示。

输入为一个二维数组 docs，docs[i] 表示 id 为 i 的文档。返回一个数组，其中每个元素是一个字符串，代表每对相似度大于 0 的文档，其格式为 {id1},{id2}: {similarity}，其中 id1 为两个文档中较小的 id，similarity 为相似度，精确到小数点后 4 位。以任意顺序返回数组均可。

提示：

1
2
docs.length <= 500  
docs[i].length <= 500

输入:
[
[14, 15, 100, 9, 3],
[32, 1, 9, 3, 5],
[15, 29, 2, 6, 8, 7],
[7, 10]
]
输出:
[
“0,1: 0.2500”,
“0,2: 0.1000”,
“2,3: 0.1429”
]

题解

算法1 : 正向索引 (超时)

用以下数据结构维护文档信息：settings[i] 保存第 i 个文档中的所有去重后的单词。

1
vector<unordered_set<int>> settings;

对每对文档 (i, j)，通过遍历 settings[i]和 settings[j]，求出交集和并集，得出相似度。

相似度压进答案时，需要注意位数限制和四舍五入的正确性，在求出 double 类型的相似度 sim 后，用以下代码转换为 string，并限制 4 位小数。

注意 1e-9。

1
2
3
4
5
string sim_str;
stringstream ss;
ss << std::fixed << std::setprecision(4);
ss << sim + 1e-9;
ss >> sim_str;

建索引时间复杂度 $O(DW)$, D 为文档数，W 为文档中的单词数；查询的时间复杂度 $O(D^{2}W)$。

代码(C++)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class Solution {
public:
    vector<string> computeSimilarities(vector<vector<int>>& docs) {
        int n = docs.size();
        settings = vector<unordered_set<int>>(n);
        for(int i = 0; i < n; ++i)
        {
            for(int a: docs[i])
                settings[i].insert(a);
        }
        vector<string> result;
        for(int i = 0; i < n - 1; ++i)
        {
            if(docs[i].empty())
                continue;
            for(int j = i + 1; j < n; ++j)
            {
                if(docs[j].empty())
                    continue;
                string res = "";
                res += to_string(i) + ',' + to_string(j) + ": ";
                double sim = get_sim(i, j);
                string sim_str;
                stringstream ss;
                ss << std::fixed << std::setprecision(4);
                ss << sim + 1e-9;
                ss >> sim_str;
                if(sim_str != "0.0000")
                {
                    res += sim_str;
                    result.push_back(res);
                }
            }
        }
        return result;
    }

private:
    vector<unordered_set<int>> settings;

    double get_sim(int i, int j)
    {
        int union_size = 0, intersection_size = 0;
        for(const int &a: settings[i])
        {
            if(settings[j].count(a) > 0)
                ++intersection_size;
            ++union_size;
        }
        for(const int &a: settings[j])
        {
            if(settings[i].count(a) == 0)
                ++union_size;
        }
        return (double)intersection_size / union_size;
    }
};

算法2 : 倒排索引

用以下数据结构维护文档信息，mapping[i] 是一个哈希集合，表示出现单词 i 的文档集合。

1
unordered_map<int, unordered_set<int>> mapping;

对每对文档 (i, j), 通过遍历 文档 i 的单词 for(int word: docs[i]), 查询 mapping[word] 是否出现在文档 j 中，如果出现，则交集的大小 + 1;

得到交集大小后，并集的大小为两个文档 docs[i], docs[j] 的总词数减去交集大小。

建索引的时间复杂度同样是 $O(DW)$，查询的时间复杂度 $O(D^{2}W)$，这个倒排索引的算法与正向索引相比，查询的时间复杂度并没有降，结果也是超时的。但是这个算法可以继续优化。

代码(C++)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class Solution {
public:
    vector<string> computeSimilarities(vector<vector<int>>& docs) {
        int n = docs.size();
        mapping = unordered_map<int, unordered_set<int>>();
        for(int i = 0; i < n; ++i)
        {
            for(int a: docs[i])
                mapping[a].insert(i);
        }
        vector<string> result;
        for(int i = 0; i < n - 1; ++i)
        {
            if(docs[i].empty())
                continue;
            for(int j = i + 1; j < n; ++j)
            {
                if(docs[j].empty())
                    continue;
                string res = "";
                res += to_string(i) + ',' + to_string(j) + ": ";
                double sim = get_sim(i, j, docs);
                string sim_str;
                stringstream ss;
                ss << std::fixed << std::setprecision(4);
                ss << sim + 1e-9;
                ss >> sim_str;
                if(sim_str != "0.0000")
                {
                    res += sim_str;
                    result.push_back(res);
                }
            }
        }
        return result;
    }

private:
    unordered_map<int, unordered_set<int>> mapping;

    double get_sim(int i, int j, const vector<vector<int>>& docs)
    {
        int union_size = docs[i].size() + docs[j].size(), intersection_size = 0;
        for(int word_i: docs[i])
        {
            if(mapping[word_i].count(j) > 0)
                ++intersection_size;
        }
        union_size -= intersection_size;
        return (double)intersection_size / union_size;
    }
};

优化

倒排索引优化

根据 mapping 我们就可以判断哪两个文档会相交，因为相交是稀疏的，所以相交的判断是很少的。

如果有一个矩阵 intersection[i][j] := 文档 i 和文档 j 的交集大小，这个矩阵就是稀疏的，即其中 D^{2} 个格子中大部分都是 0。

前面的算法，查询的时间复杂度是 $O(D^{2}W)$，每次查询是 $O(W)$。如果可以以小于 $O(D^{2}W)$ 的时间复杂度将 intersection 这个矩阵预处理出来，那么每次查询的时间复杂度就可以摊销小于 $O(W)$，这就是利用倒排索引的优化点。

预处理过程如下：

1
2
3
枚举 mapping 中的每个词 word:
    枚举 mapping[word] 中的文档对 (i, j):
        ++intersection[i][j]

有了预处理出来的 intersection, 相似度查询可以查表，如果 intersection[i][j] == 0, 直接跳过，如果 intersection[i][j] != 0, 相似度公式：intersection[i][j] / (docs[i].size() + docs[j].size() - intersection[i][j])

建索引的时间复杂度同样是 $O(DW)$，查询的时间复杂度 $O(1)$，多了一个预处理过程，预处理过程的时间复杂度为 $O(N^{2}W)$，其中 N 为单词出现的文档个数。因为文档的相交是稀疏的，所以 $N << D$，因此 $O(N^{2}W) << O(D^{2}W)$。每次查询的时间复杂度为摊销 $O((\frac{N}{D})^{2}W)$

由于 mapping 只用于预处理，查询阶段不用再去查找了，因此维护文档信息的数据结构可以改为：

1
unordered_map<int, vector<int>> mapping;

intersection 的数据结构为 vector<vector<int>>

输入输出优化

将结果输出到答案的部分：string + stringstream 改成 char[256] + sprintf，实测可以加速 30%。

注意两种写法在将 double 传入的时候都要加 1e-9。

1
2
3
4
5
6
7
8
9
10
#include <sstream>
string res = "";
res += to_string(i) + ',' + to_string(j) + ": ";
string sim_str;
stringstream ss;
ss << std::fixed << std::setprecision(4);
ss << sim + 1e-9;
ss >> sim_str;
res += sim_str;
result.push_back(res);
1
2
3
4
#include <cstdio>
char res[1024];
sprintf(res, "%d,%d: %0.4lf", i, j, sim + 1e-9);
result.push_back(res);

代码(C++)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class Solution {
public:
    vector<string> computeSimilarities(vector<vector<int>>& docs) {
        int n = docs.size();
        mapping = unordered_map<int, vector<int>>();
        intersection = vector<vector<int>>(n, vector<int>(n));
        for(int i = 0; i < n; ++i)
        {
            for(int a: docs[i])
                mapping[a].push_back(i);
        }
        for(const auto& word_info: mapping)
        {
            int m = (word_info.second).size();
            for(int i = 0; i < m - 1; ++i)
                for(int j = 0; j < m; ++j)
                {
                    int x = word_info.second[i];
                    int y = word_info.second[j];
                    ++intersection[x][y];
                }
        }
        vector<string> result;
        for(int i = 0; i < n - 1; ++i)
        {
            if(docs[i].empty())
                continue;
            for(int j = i + 1; j < n; ++j)
            {
                if(docs[j].empty())
                    continue;
                if(intersection[i][j] == 0)
                    continue;
                double sim = (double)intersection[i][j] / (docs[i].size() + docs[j].size() - intersection[i][j]);
                if(sim + 1e9 < 0.00005)
                    continue;
                // string res = "";
                // res += to_string(i) + ',' + to_string(j) + ": ";
                // string sim_str;
                // stringstream ss;
                // ss << std::fixed << std::setprecision(4);
                // ss << sim + 1e-9;
                // ss >> sim_str;
                // res += sim_str;
                char res[1024];
                sprintf(res, "%d,%d: %0.4lf", i, j, sim + 1e-9);
                result.push_back(res);
            }
        }
        return result;
    }

private:
    unordered_map<int, vector<int>> mapping;
    vector<vector<int>> intersection;
};
Share