Python:基于交叉点的简单列表合并

Question

考虑有一些整数列表:

#--------------------------------------
0 [0,1,3]
1 [1,0,3,4,5,10,...]
2 [2,8]
3 [3,1,0,...]
...
n []
#--------------------------------------

问题是合并具有至少一个共同元素的列表.因此,仅给定部分的结果如下:

#--------------------------------------
0 [0,1,3,4,5,10,...]
2 [2,8]
#--------------------------------------

在大数据上执行此操作的最有效方法是什么(元素只是数字)？ 是tree结构一些思考？我现在通过将列表转换为sets迭代并迭代交叉来完成工作,但它很慢!而且我有一种如此初级的感觉!此外,实现缺少一些东西(未知),因为有些列表有时会保持未合并!话虽如此,如果你提出自我实现,请慷慨并提供一个简单的示例代码[显然Python是我的偏爱:)]或pesudo代码.
更新1: 这是我使用的代码:

#--------------------------------------
lsts = [[0,1,3],
        [1,0,3,4,5,10,11],
        [2,8],
        [3,1,0,16]];
#--------------------------------------

功能是(越野车!!):

#--------------------------------------
def merge(lsts):
    sts = [set(l) for l in lsts]
    i = 0
    while i < len(sts):
        j = i+1
        while j < len(sts):
            if len(sts[i].intersection(sts[j])) > 0:
                sts[i] = sts[i].union(sts[j])
                sts.pop(j)
            else: j += 1                        #---corrected
        i += 1
    lst = [list(s) for s in sts]
    return lst
#--------------------------------------

结果是:

#--------------------------------------
>>> merge(lsts)
>>> [0, 1, 3, 4, 5, 10, 11, 16], [8, 2]]
#--------------------------------------

更新2:根据 我的经验,下面的Niklas Baumstark给出的代码显示对于简单的情况来说要快一点.尚未测试"Hooked"给出的方法,因为它是完全不同的方法(顺便说一下它似乎很有趣).所有这些的测试程序可能非常难以或无法确保结果.我将使用的真实数据集是如此之大和复杂,因此不可能仅通过重复来追踪任何错误.这就是我需要100%满足方法的可靠性,然后才能将它作为模块在大型代码中推送到位.简单来说,尼克拉斯的方法更快,简单的答案当然是正确的.
但是,我如何确定它适用于真正的大型数据集？因为我无法直观地追踪错误!

更新3: 请注意,该方法的可靠性比此问题的速度更重要.我希望能够最终将Python代码转换为Fortran以获得最佳性能.

更新4:
这篇文章中有许多有趣的观点,并慷慨地给出答案,建设性意见.我建议你仔细阅读.请接受我对这个问题的发展,惊人的答案以及建设性的评论和讨论的赞赏.

Answer 1

我的尝试:

def merge(lsts):
    sets = [set(lst) for lst in lsts if lst]
    merged = True
    while merged:
        merged = False
        results = []
        while sets:
            common, rest = sets[0], sets[1:]
            sets = []
            for x in rest:
                if x.isdisjoint(common):
                    sets.append(x)
                else:
                    merged = True
                    common |= x
            results.append(common)
        sets = results
    return sets

lst = [[65, 17, 5, 30, 79, 56, 48, 62],
       [6, 97, 32, 93, 55, 14, 70, 32],
       [75, 37, 83, 34, 9, 19, 14, 64],
       [43, 71],
       [],
       [89, 49, 1, 30, 28, 3, 63],
       [35, 21, 68, 94, 57, 94, 9, 3],
       [16],
       [29, 9, 97, 43],
       [17, 63, 24]]
print merge(lst)

基准测试:

import random

# adapt parameters to your own usage scenario
class_count = 50
class_size = 1000
list_count_per_class = 100
large_list_sizes = list(range(100, 1000))
small_list_sizes = list(range(0, 100))
large_list_probability = 0.5

if False:  # change to true to generate the test data file (takes a while)
    with open("/tmp/test.txt", "w") as f:
        lists = []
        classes = [
            range(class_size * i, class_size * (i + 1)) for i in range(class_count)
        ]
        for c in classes:
            # distribute each class across ~300 lists
            for i in xrange(list_count_per_class):
                lst = []
                if random.random() < large_list_probability:
                    size = random.choice(large_list_sizes)
                else:
                    size = random.choice(small_list_sizes)
                nums = set(c)
                for j in xrange(size):
                    x = random.choice(list(nums))
                    lst.append(x)
                    nums.remove(x)
                random.shuffle(lst)
                lists.append(lst)
        random.shuffle(lists)
        for lst in lists:
            f.write(" ".join(str(x) for x in lst) + "\n")

setup = """
# Niklas'
def merge_niklas(lsts):
    sets = [set(lst) for lst in lsts if lst]
    merged = 1
    while merged:
        merged = 0
        results = []
        while sets:
            common, rest = sets[0], sets[1:]
            sets = []
            for x in rest:
                if x.isdisjoint(common):
                    sets.append(x)
                else:
                    merged = 1
                    common |= x
            results.append(common)
        sets = results
    return sets

# Rik's
def merge_rik(data):
    sets = (set(e) for e in data if e)
    results = [next(sets)]
    for e_set in sets:
        to_update = []
        for i, res in enumerate(results):
            if not e_set.isdisjoint(res):
                to_update.insert(0, i)

        if not to_update:
            results.append(e_set)
        else:
            last = results[to_update.pop(-1)]
            for i in to_update:
                last |= results[i]
                del results[i]
            last |= e_set
    return results

# katrielalex's
def pairs(lst):
    i = iter(lst)
    first = prev = item = i.next()
    for item in i:
        yield prev, item
        prev = item
    yield item, first

import networkx

def merge_katrielalex(lsts):
    g = networkx.Graph()
    for lst in lsts:
        for edge in pairs(lst):
            g.add_edge(*edge)
    return networkx.connected_components(g)

# agf's (optimized)
from collections import deque

def merge_agf_optimized(lists):
    sets = deque(set(lst) for lst in lists if lst)
    results = []
    disjoint = 0
    current = sets.pop()
    while True:
        merged = False
        newsets = deque()
        for _ in xrange(disjoint, len(sets)):
            this = sets.pop()
            if not current.isdisjoint(this):
                current.update(this)
                merged = True
                disjoint = 0
            else:
                newsets.append(this)
                disjoint += 1
        if sets:
            newsets.extendleft(sets)
        if not merged:
            results.append(current)
            try:
                current = newsets.pop()
            except IndexError:
                break
            disjoint = 0
        sets = newsets
    return results

# agf's (simple)
def merge_agf_simple(lists):
    newsets, sets = [set(lst) for lst in lists if lst], []
    while len(sets) != len(newsets):
        sets, newsets = newsets, []
        for aset in sets:
            for eachset in newsets:
                if not aset.isdisjoint(eachset):
                    eachset.update(aset)
                    break
            else:
                newsets.append(aset)
    return newsets

# alexis'
def merge_alexis(data):
    bins = range(len(data))  # Initialize each bin[n] == n
    nums = dict()

    data = [set(m) for m in data]  # Convert to sets
    for r, row in enumerate(data):
        for num in row:
            if num not in nums:
                # New number: tag it with a pointer to this row's bin
                nums[num] = r
                continue
            else:
                dest = locatebin(bins, nums[num])
                if dest == r:
                    continue  # already in the same bin

                if dest > r:
                    dest, r = r, dest  # always merge into the smallest bin

                data[dest].update(data[r])
                data[r] = None
                # Update our indices to reflect the move
                bins[r] = dest
                r = dest

    # Filter out the empty bins
    have = [m for m in data if m]
    return have

def locatebin(bins, n):
    while bins[n] != n:
        n = bins[n]
    return n

lsts = []
size = 0
num = 0
max = 0
for line in open("/tmp/test.txt", "r"):
    lst = [int(x) for x in line.split()]
    size += len(lst)
    if len(lst) > max:
        max = len(lst)
    num += 1
    lsts.append(lst)
"""

setup += """
print "%i lists, {class_count} equally distributed classes, average size %i, max size %i" % (num, size/num, max)
""".format(class_count=class_count)

import timeit
print "niklas"
print timeit.timeit("merge_niklas(lsts)", setup=setup, number=3)
print "rik"
print timeit.timeit("merge_rik(lsts)", setup=setup, number=3)
print "katrielalex"
print timeit.timeit("merge_katrielalex(lsts)", setup=setup, number=3)
print "agf (1)"
print timeit.timeit("merge_agf_optimized(lsts)", setup=setup, number=3)
print "agf (2)"
print timeit.timeit("merge_agf_simple(lsts)", setup=setup, number=3)
print "alexis"
print timeit.timeit("merge_alexis(lsts)", setup=setup, number=3)

这些时间显然取决于基准测试的具体参数,例如类的数量,列表的数量,列表大小等.根据您的需要调整这些参数以获得更有用的结果.

以下是我的机器上针对不同参数的一些示例输出.他们表明所有算法都有自己的优点和缺点,具体取决于他们得到的输入类型:

=====================
# many disjoint classes, large lists
class_count = 50
class_size = 1000
list_count_per_class = 100
large_list_sizes = list(range(100, 1000))
small_list_sizes = list(range(0, 100))
large_list_probability = 0.5
=====================

niklas
5000 lists, 50 equally distributed classes, average size 298, max size 999
4.80084705353
rik
5000 lists, 50 equally distributed classes, average size 298, max size 999
9.49251699448
katrielalex
5000 lists, 50 equally distributed classes, average size 298, max size 999
21.5317108631
agf (1)
5000 lists, 50 equally distributed classes, average size 298, max size 999
8.61671280861
agf (2)
5000 lists, 50 equally distributed classes, average size 298, max size 999
5.18117713928
=> alexis
=> 5000 lists, 50 equally distributed classes, average size 298, max size 999
=> 3.73504281044

===================
# less number of classes, large lists
class_count = 15
class_size = 1000
list_count_per_class = 300
large_list_sizes = list(range(100, 1000))
small_list_sizes = list(range(0, 100))
large_list_probability = 0.5
===================

niklas
4500 lists, 15 equally distributed classes, average size 296, max size 999
1.79993700981
rik
4500 lists, 15 equally distributed classes, average size 296, max size 999
2.58237695694
katrielalex
4500 lists, 15 equally distributed classes, average size 296, max size 999
19.5465381145
agf (1)
4500 lists, 15 equally distributed classes, average size 296, max size 999
2.75445604324
=> agf (2)
=> 4500 lists, 15 equally distributed classes, average size 296, max size 999
=> 1.77850699425
alexis
4500 lists, 15 equally distributed classes, average size 296, max size 999
3.23530197144

===================
# less number of classes, smaller lists
class_count = 15
class_size = 1000
list_count_per_class = 300
large_list_sizes = list(range(100, 1000))
small_list_sizes = list(range(0, 100))
large_list_probability = 0.1
===================

niklas
4500 lists, 15 equally distributed classes, average size 95, max size 997
0.773697137833
rik
4500 lists, 15 equally distributed classes, average size 95, max size 997
1.0523750782
katrielalex
4500 lists, 15 equally distributed classes, average size 95, max size 997
6.04466891289
agf (1)
4500 lists, 15 equally distributed classes, average size 95, max size 997
1.20285701752
=> agf (2)
=> 4500 lists, 15 equally distributed classes, average size 95, max size 997
=> 0.714507102966
alexis
4500 lists, 15 equally distributed classes, average size 95, max size 997
1.1286110878

Answer 2

我试图在这个问题和重复的问题中总结所有关于这个主题的说法和做法.

我试图测试和时间的每一个解决方案(所有的代码在这里).

测试

这是TestCase来自测试模块:

class MergeTestCase(unittest.TestCase):

    def setUp(self):
        with open('./lists/test_list.txt') as f:
            self.lsts = json.loads(f.read())
        self.merged = self.merge_func(deepcopy(self.lsts))

    def test_disjoint(self):
        """Check disjoint-ness of merged results"""
        from itertools import combinations
        for a,b in combinations(self.merged, 2):
            self.assertTrue(a.isdisjoint(b))

    def test_coverage(self):    # Credit to katrielalex
        """Check coverage original data"""
        merged_flat = set()
        for s in self.merged:
            merged_flat |= s

        original_flat = set()
        for lst in self.lsts:
            original_flat |= set(lst)

        self.assertTrue(merged_flat == original_flat)

    def test_subset(self):      # Credit to WolframH
        """Check that every original data is a subset"""
        for lst in self.lsts:
            self.assertTrue(any(set(lst) <= e for e in self.merged))

这个测试假设了一个集合列表,因此我无法测试一些与列表一起使用的漏洞.

我无法测试以下内容:

katrielalex
steabert

在我可以测试的那些中,有两个失败了:

  -- Going to test: agf (optimized) --
Check disjoint-ness of merged results ... FAIL

  -- Going to test: robert king --
Check disjoint-ness of merged results ... FAIL

定时

性能与所采用的数据测试密切相关.

到目前为止,有三个答案试图为他们和他人解决问题.由于他们使用不同的测试数据,他们有不同的结果

1. 尼克拉斯的基准测试非常好.用他的banchmark可以做一些不同的测试来改变一些参数.

   我使用了他在自己的答案中使用的相同的三组参数,并将它们放在三个不同的文件中:

   filename = './lists/timing_1.txt'
   class_count = 50,
   class_size = 1000,
   list_count_per_class = 100,
   large_list_sizes = (100, 1000),
   small_list_sizes = (0, 100),
   large_list_probability = 0.5,
   
   filename = './lists/timing_2.txt'
   class_count = 15,
   class_size = 1000,
   list_count_per_class = 300,
   large_list_sizes = (100, 1000),
   small_list_sizes = (0, 100),
   large_list_probability = 0.5,
   
   filename = './lists/timing_3.txt'
   class_count = 15,
   class_size = 1000,
   list_count_per_class = 300,
   large_list_sizes = (100, 1000),
   small_list_sizes = (0, 100),
   large_list_probability = 0.1,
   

   这是我得到的结果:

   从文件: timing_1.txt

   Timing with: >> Niklas << Benchmark
   Info: 5000 lists, average size 305, max size 999
   
   Timing Results:
   10.434  -- alexis
   11.476  -- agf
   11.555  -- Niklas B.
   13.622  -- Rik. Poggi
   14.016  -- agf (optimized)
   14.057  -- ChessMaster
   20.208  -- katrielalex
   21.697  -- steabert
   25.101  -- robert king
   76.870  -- Sven Marnach
   133.399  -- hochl
   

   从文件: timing_2.txt

   Timing with: >> Niklas << Benchmark
   Info: 4500 lists, average size 305, max size 999
   
   Timing Results:
   8.247  -- Niklas B.
   8.286  -- agf
   8.637  -- Rik. Poggi
   8.967  -- alexis
   9.090  -- ChessMaster
   9.091  -- agf (optimized)
   18.186  -- katrielalex
   19.543  -- steabert
   22.852  -- robert king
   70.486  -- Sven Marnach
   104.405  -- hochl
   

   从文件: timing_3.txt

   Timing with: >> Niklas << Benchmark
   Info: 4500 lists, average size 98, max size 999
   
   Timing Results:
   2.746  -- agf
   2.850  -- Niklas B.
   2.887  -- Rik. Poggi
   2.972  -- alexis
   3.077  -- ChessMaster
   3.174  -- agf (optimized)
   5.811  -- katrielalex
   7.208  -- robert king
   9.193  -- steabert
   23.536  -- Sven Marnach
   37.436  -- hochl
   

2. 通过Sven的测试数据,我得到了以下结果:

   Timing with: >> Sven << Benchmark
   Info: 200 lists, average size 10, max size 10
   
   Timing Results:
   2.053  -- alexis
   2.199  -- ChessMaster
   2.410  -- agf (optimized)
   3.394  -- agf
   3.398  -- Rik. Poggi
   3.640  -- robert king
   3.719  -- steabert
   3.776  -- Niklas B.
   3.888  -- hochl
   4.610  -- Sven Marnach
   5.018  -- katrielalex
   

3. 最后,根据Agf的基准,我得到了:

   Timing with: >> Agf << Benchmark
   Info: 2000 lists, average size 246, max size 500
   
   Timing Results:
   3.446  -- Rik. Poggi
   3.500  -- ChessMaster
   3.520  -- agf (optimized)
   3.527  -- Niklas B.
   3.527  -- agf
   3.902  -- hochl
   5.080  -- alexis
   15.997  -- steabert
   16.422  -- katrielalex
   18.317  -- robert king
   1257.152  -- Sven Marnach
   

正如我在开始时所说的,所有代码都可以在这个git存储库中找到.所有合并函数都在一个名为的文件中core.py,其名称以其结尾的每个函数_merge都将在测试期间自动加载,因此添加/测试/改进自己的解决方案应该不难.

让我也知道是否有什么问题,这是很多编码,我可以使用一些新鲜的眼睛:)

Answer 3

使用矩阵操作

让我以下面的评论为这个答案作序:

这是错误的做法.它是数字不稳定的,并且比其他方法更快,自行承担使用风险.

话虽如此,我无法抗拒从动态角度解决问题(我希望你能对这个问题有一个全新的看法).从理论上讲,这应该始终有效,但特征值计算通常会失败.我们的想法是将您的列表视为从行到列的流.如果两行共享一个公共值,则它们之间存在连接流.如果我们将这些流量视为水,我们会看到流量在它们之间存在连接路径时聚集成小池.为简单起见,我将使用较小的集合,但它也适用于您的数据集:

from numpy import where, newaxis
from scipy import linalg, array, zeros

X = [[0,1,3],[2],[3,1]]

我们需要将数据转换为流程图.如果第i行流入值j,我们将其放入矩阵中.这里我们有3行和4个唯一值:

A = zeros((4,len(X)), dtype=float)
for i,row in enumerate(X):
    for val in row: A[val,i] = 1

通常,您需要更改4以捕获您拥有的唯一值的数量.如果该集合是一个从0开始的整数列表,您可以简单地将其设为最大数字.我们现在执行特征值分解.准确的SVD,因为我们的矩阵不是正方形.

S  = linalg.svd(A)

我们希望仅保留此答案的3x3部分,因为它将代表池的流量.实际上我们只想要这个矩阵的绝对值; 我们只关心这个集群空间中是否有流.

M  = abs(S[2])

我们可以将这个矩阵M看作马尔可夫矩阵,并通过行规范化使其明确.一旦我们有了这个,我们计算(左)特征值分解.这个矩阵.

M /=  M.sum(axis=1)[:,newaxis]
U,V = linalg.eig(M,left=True, right=False)
V = abs(V)

现在,断开的(非遍历)马尔可夫矩阵具有良好的性质,对于每个非连通的簇,存在统一的特征值.与这些统一值相关联的特征向量是我们想要的:

idx = where(U > .999)[0]
C = V.T[idx] > 0

由于上述数值不稳定,我必须使用.999.在这一点上,我们完成了!现在,每个独立集群都可以拉出相应的行:

for cluster in C:
    print where(A[:,cluster].sum(axis=1))[0]

这按预期给出:

[0 1 3]
[2]

换X到你的lst,你会得到:[ 0 1 3 4 5 10 11 16] [2 8].

附录

为什么这有用呢？我不知道您的基础数据来自哪里,但是当连接不是绝对的时会发生什么？Say row 80%的时间1都有条目3- 您如何概括问题？上面的流程方法可以正常工作,并且可以通过该.999值进行完全参数化,距离统一越远,关联就越松散.

视觉表现

由于图片值1K字,下面是我的例子和你的矩阵A和V的图lst.注意如何V分成两个簇(它是一个块对角矩阵,在置换后有两个块),因为每个例子只有两个唯一列表!

更快的实施

事后看来,我意识到你可以跳过SVD步骤并只计算一次分解:

M = dot(A.T,A)
M /=  M.sum(axis=1)[:,newaxis]
U,V = linalg.eig(M,left=True, right=False)

这种方法(除了速度)的优点M是现在是对称的,因此计算可以更快,更准确(无需担心的虚数值).

Answer 4

编辑:好的,其他问题已经关闭,发布在这里.

好问题!如果您将其视为图中的连接组件问题,则会更简单.以下代码使用优秀的networkx图形库和此问题的pairs函数.

def pairs(lst):
    i = iter(lst)
    first = prev = item = i.next()
    for item in i:
        yield prev, item
        prev = item
    yield item, first

lists = [[1,2,3],[3,5,6],[8,9,10],[11,12,13]]

import networkx
g = networkx.Graph()
for sub_list in lists:
    for edge in pairs(sub_list):
            g.add_edge(*edge)

networkx.connected_components(g)
[[1, 2, 3, 5, 6], [8, 9, 10], [11, 12, 13]]

说明

我们创建一个新的(空)图g.对于每个子列表lists,将其元素视为图的节点,并在它们之间添加边.(因为我们只关心连通性,所以我们不需要添加所有边 - 只有相邻的边!)注意,add_edge需要两个对象,将它们视为节点(如果它们不存在则添加它们),并添加他们之间的边缘.

然后,我们只是找到图表的连接组件 - 一个已解决的问题! - 并将它们作为交叉集输出.

Answer 5

这是我的答案。我还没有根据今天的一批答案进行检查。

基于交集的算法是 O(N^2)，因为它们根据所有现有的检查每个新集合，所以我使用了一种方法来索引每个数字并运行接近 O(N)（如果我们接受字典查找是O(1))。然后我运行了基准测试，感觉自己像个彻头彻尾的白痴，因为它跑得更慢，但仔细检查后发现测试数据最终只有少数不同的结果集，所以二次算法没有太多工作要做做。用超过 10-15 个不同的 bin 测试它，我的算法要快得多。尝试使用 50 多个不同 bin 的测试数据，它的速度要快得多。

（编辑：基准测试的运行方式也存在问题，但我的诊断是错误的。我修改了代码以适应重复测试的运行方式）。

def mergelists5(data):
    """Check each number in our arrays only once, merging when we find
    a number we have seen before.
    """

    bins = range(len(data))  # Initialize each bin[n] == n
    nums = dict()

    data = [set(m) for m in data ]  # Convert to sets    
    for r, row in enumerate(data):
        for num in row:
            if num not in nums:
                # New number: tag it with a pointer to this row's bin
                nums[num] = r
                continue
            else:
                dest = locatebin(bins, nums[num])
                if dest == r:
                    continue # already in the same bin

                if dest > r:
                    dest, r = r, dest   # always merge into the smallest bin

                data[dest].update(data[r]) 
                data[r] = None
                # Update our indices to reflect the move
                bins[r] = dest
                r = dest 

    # Filter out the empty bins
    have = [ m for m in data if m ]
    print len(have), "groups in result"
    return have


def locatebin(bins, n):
    """
    Find the bin where list n has ended up: Follow bin references until
    we find a bin that has not moved.
    """
    while bins[n] != n:
        n = bins[n]
    return n