基于集合的三元组的高效匹配算法

Question

我正在寻找一种有效的方法来解决以下问题。

列表 1 是由原始三元组标识的记录列表：

X | Y | Z

列表 2 是由三个集合标识的记录列表。一个 X，一个 Y，一个 Z。 X、Y、Z 与列表一中的 X、Y、Z 具有相同的“类型”，因此可以直接相互比较。

Set(X) | Set(Y) | Set(Z)

对于列表 1 中的某个项目，我需要查找列表 2 中的所有项目，其中列表 1 中的 X、Y、Z 全部出现在列表 2 中相应的集合中。这可以通过示例进行最佳演示：

列表 1：

X1, Y1, Z1

列表 2：

(X1, X2) | (Y1) | (Z1, Z3)

(X1) | (Y1, Y2) | (Z1, Z2, Z3)

(X3) | (Y1, Y3) | (Z2, Z3)

在上面，列表 1 中的项目将匹配列表 2 中的前两项。第三个项目将不会匹配，因为 X1 没有出现在 X 集合中，并且 Z1 不会出现在 Z 集合中。

我已经编写了该算法的功能正确版本，但担心较大数据集上的性能。两个列表都非常大，因此迭代列表 1，然后对列表 2 的每个项目执行迭代将非常低效。

我尝试通过将列表 2 中的每个项目反规范化为一个映射来构建索引，但每个项目的索引中的索引条目数与该项目子集的大小成正比。因此，这使用了非常高的内存级别，并且还需要一些重要的资源来构建。

谁能向我建议解决这个问题的最佳方法。我很乐意考虑内存和 CPU 的最佳解决方案，但取得平衡会很好！

Answer 1

如果集合的总大小不太大，您可以尝试将列表 2 建模为位字段。不过，该结构可能会非常分散 - 可能是维基百科文章中关于位数组中引用的结构（Judy 数组、tries、布隆过滤器）可以帮助解决规范化方法的内存问题。

Answer 2

您可以从 List2 构建一棵树；树的第一层是集合 X 中出现的 (X1..Xn) 的第一个。第二层是第二项的值，加上包含仅包含 X1 的列表集的叶节点。下一个级别包含下一个可能的值，依此类推。

Root --+--X1--+--EOF--> List of pointers to list2 lines containing only "X1"
       |      |
       |      +--X2---+--EOF--> List of pointers to list2 lines containing only "X1,X2"
       |      |       |
       |      |       +--X3--+--etc--
       |      |       
       |      +--X3---+--EOF--> "X1,X3"
       |             
       +--X2--+--EOF--> "X2"
       |      |
       |      +--X3---+--EOF--> "X2,X3"
       |      |       |
       ...

这在内存消耗上是昂贵的（我认为 N^2 log K？其中 N=X 的值，K=List2 中的行），但会导致快速检索时间。如果可能的 X 数量很大，那么这种方法就会失败......

显然，您可以为元组的所有 3 个部分构建此索引，然后将搜索每棵树的结果 AND 在一起。

Answer 3

有一种相当有效的方法可以通过单次传递 list2 来完成此操作。首先构建 list1 中项目的索引。

from collections import defaultdict

# index is HashMap<X, HashMap<Y, HashMap<Z, Integer>>>
index = defaultdict(lambda: defaultdict(dict))
for rowid, (x, y, z) in enumerate(list1):
    index[x][y][z] = rowid

for rowid2, (xs, ys, zs) in enumerate(list2):
    xhits = defaultdict(list)
    for x in xs:
        if x in index:
            for y, zmap in index[x].iteritems():
                xhits[y].append(zmap)

    yhits = defaultdict(list)
    for y in ys:
        if y in xhits:
            for z, rowid1 in xhits[y].iteritems():
                yhits[z].append(rowid1)

    for z in zs:
        if z in yhits:
            for rowid1 in yhits[z]:
                print "list1[%d] matches list2[%d]" % (hit[z], rowid2)

这里额外的簿记可能会使其比索引 list2 慢。但由于在您的情况下 list1 通常比 list2 小得多，因此这将使用更少的内存。如果您从磁盘读取 list2，则使用此算法您永远不需要将其任何部分保留在内存中。

内存访问可能是一件大事，所以我不能肯定地说哪个在实践中会更快。必须测量。除非哈希表出现故障，这两种情况下最坏情况的时间复杂度均为 O(len(list1)*len(list2))。

Answer 4

对于 List 2 使用 HashSet （或 HashSets）怎么样？这样您只需要迭代列表 1

Answer 5

如果您使用 Guava，则有一种高级方法可以实现此目的，但不一定是 最佳但不会做任何疯狂的事情：

List<SomeType> list1 = ...;
List<Set<SomeType>> candidateFromList2 = ...;
if (Sets.cartesianProduct(candidateFromList2).contains(list1)) { ... }

但检查这个“常规”也不难。

Answer 6

将会有很多方法来解决这个问题。哪个是正确的取决于数据和可用内存的大小。

一种简单的技术是从 list2 构建一个表，以加速来自 list1 的查询。

from collections import defaultdict

# Build "hits".  hits[0] is a table of, for each x,
# which items in list2 contain it. Likewise hits[1]
# is for y and hits[2] is for z.
hits = [defaultdict(set) for i in range(3)]
for rowid, row in enumerate(list2):
    for i in range(3):
        for v in row[i]:
            hits[i][v].add(rowid)

# For each row, query the database to find which
# items in list2 contain all three values.
for x, y, z in list1:
    print hits[0][x].intersection(hits[1][y], hits[2][z])