如何使用多处理在一个非常大的列表中删除重复项？

Question

比如说，我有一个包含随机数的巨大列表

L = [random.randrange(0,25000000000) for _ in range(1000000000)]

我需要去掉这个列表中的重复项

我为包含较少元素的列表编写了这段代码

def remove_duplicates(list_to_deduplicate):
seen = set()
result=[]
for i in list_to_deduplicate:
    if i not in seen:
        result.append(i)
        seen.add(i)
return result

在上面的代码中，我创建了一个集合，这样我就可以记住我正在处理的列表中已经出现的数字。如果该数字不在集合中，那么我将其添加到结果列表中，我需要返回并将其保存在集合中，这样它就不会再次添加到结果列表中

现在对于列表中的1000000个数字，一切都很好，我可以很快得到结果，但是对于优于100000000的数字，比如说出现100000000个问题，我需要在我的机器上使用不同的内核来尝试解决问题，然后合并多个进程的结果

我的第一个猜测是让所有流程都可以访问一个集合，但会出现许多复杂情况 一个进程如何读取，而另一个进程正在添加到集合中，我甚至不知道是否可以在进程之间共享一个集合我知道我们可以使用队列或管道，但我不确定如何使用它

有人能给我一个建议，什么是解决这个问题的最好方法 我愿意接受任何新想法

Answer 1

我怀疑即使是你最棒的列表也足够大，以至于多处理可以改善计时。使用numpy和多线程可能是最好的选择

多处理引入了相当多的开销并增加了内存消耗，就像前面正确提到的@Frank Merrow一样。 不过，对于多线程来说，情况并非如此。重要的是不要混淆这些术语，因为进程和线程是不同的。 同一进程内的线程共享内存，不同进程则不共享

在Python中使用多核的问题是GIL，它不允许多个线程（在同一进程中）并行执行Python字节码。一些C扩展（如numpy）可以释放GIL，这可以通过多线程从多核并行中获益。在这里，您有机会通过使用numpy在巨大改进的基础上获得一些加速

from multiprocessing.dummy import Pool  # .dummy uses threads
import numpy as np

r = np.random.RandomState(42).randint(0, 25000000000, 100_000_000)
n_threads = 8

result = np.unique(np.concatenate(
    Pool(n_threads).map(np.unique, np.array_split(r, n_threads)))
).tolist()

使用numpy和线程池，拆分数组，使子数组在单独的线程中唯一，然后连接子数组并使重新组合的数组再次唯一。 由于在子阵列中只能识别本地重复项，因此有必要最终删除重组阵列的重复项

对于低熵数据（许多重复项），使用^{}代替numpy.unique可以快得多。与^{}不同，它还保留了外观的顺序

请注意，只有当numpy函数通过调用低级数学库而不是多线程时，使用上述线程池才有意义。所以，总是测试它是否真的提高了性能，不要想当然

---

在以下范围内使用100M随机生成的整数进行测试：

• 高熵：0-25_000_000_000（199560个副本）
• 低熵：0-1000

代码

import time
import timeit
from multiprocessing.dummy import Pool  # .dummy uses threads

import numpy as np
import pandas as pd


def time_stmt(stmt, title=None):
    t = timeit.repeat(
        stmt=stmt,
        timer=time.perf_counter_ns, repeat=3, number=1, globals=globals()
    )
    print(f"\t{title or stmt}")
    print(f"\t\t{min(t) / 1e9:.2f} s")


if __name__ == '__main__':

    n_threads = 8  # machine with 8 cores (4 physical cores)

    stmt_np_unique_pool = \
"""
np.unique(np.concatenate(
    Pool(n_threads).map(np.unique, np.array_split(r, n_threads)))
).tolist()
"""

    stmt_pd_unique_pool = \
"""
pd.unique(np.concatenate(
    Pool(n_threads).map(pd.unique, np.array_split(r, n_threads)))
).tolist()
"""
    #                                     -

    print(f"\nhigh entropy (few duplicates) {'-' * 30}\n")
    r = np.random.RandomState(42).randint(0, 25000000000, 100_000_000)

    r = list(r)
    time_stmt("list(set(r))")

    r = np.asarray(r)
    # numpy.unique
    time_stmt("np.unique(r).tolist()")
    # pandas.unique
    time_stmt("pd.unique(r).tolist()")    
    # numpy.unique & Pool
    time_stmt(stmt_np_unique_pool, "numpy.unique() & Pool")
    # pandas.unique & Pool
    time_stmt(stmt_pd_unique_pool, "pandas.unique() & Pool")

    #  -
    print(f"\nlow entropy (many duplicates) {'-' * 30}\n")
    r = np.random.RandomState(42).randint(0, 1000, 100_000_000)

    r = list(r)
    time_stmt("list(set(r))")

    r = np.asarray(r)
    # numpy.unique
    time_stmt("np.unique(r).tolist()")
    # pandas.unique
    time_stmt("pd.unique(r).tolist()")
    # numpy.unique & Pool
    time_stmt(stmt_np_unique_pool, "numpy.unique() & Pool")
    # pandas.unique() & Pool
    time_stmt(stmt_pd_unique_pool, "pandas.unique() & Pool")

正如您在下面的计时中所看到的，仅使用numpy而不使用多线程已经是最大的性能改进。还请注意pandas.unique()对于许多重复项来说比numpy.unique()（仅）快

high entropy (few duplicates)                

    list(set(r))
        32.76 s
    np.unique(r).tolist()
        12.32 s
    pd.unique(r).tolist()
        23.01 s
    numpy.unique() & Pool
        9.75 s
    pandas.unique() & Pool
        28.91 s

low entropy (many duplicates)                

    list(set(r))
        5.66 s
    np.unique(r).tolist()
        4.59 s
    pd.unique(r).tolist()
        0.75 s
    numpy.unique() & Pool
        1.17 s
    pandas.unique() & Pool
        0.19 s