python - 使用 python 并行化在图上进行大处理

Question

我正在研究复杂网络的图表和大数据集。我使用 ndlib 库对它们运行 SIR 算法。但每次迭代大约需要 1 秒，并且代码需要 10-12 小时才能完成。我想知道有什么办法可以让它并行化吗？代码如下所示

这行代码是核心:

sir = model.infected_SIR_MODEL(it, infectionList, False)

有什么简单的方法可以让它在多线程或并行上运行吗？

count = 500
for i in numpy.arange(1, count, 1):

    for it in model.get_nodes():

        sir = model.infected_SIR_MODEL(it, infectionList, False)

每次迭代:

 for u in self.graph.nodes():

            u_status = self.status[u]
            eventp = np.random.random_sample()
            neighbors = self.graph.neighbors(u)
            if isinstance(self.graph, nx.DiGraph):
                neighbors = self.graph.predecessors(u)

            if u_status == 0:
                infected_neighbors = len([v for v in neighbors if self.status[v] == 1])
                if eventp < self.BetaList[u] * infected_neighbors:
                    actual_status[u] = 1
            elif u_status == 1:
                if eventp < self.params['model']['gamma']:
                    actual_status[u] = 2

Answer 1

因此，如果迭代是独立的，那么我看不到迭代超过 count=500 的意义。无论哪种方式multiprocessing您可能会对图书馆感兴趣。

我准备了 2 个 stub 解决方案(即根据您的具体需求进行更改)。第一个期望每个输入都是静态的(据我所知，解决方案的变化是从每次迭代内的随机状态生成中提出的)。对于第二个，您可以在 i 迭代之间更新输入数据。我没有尝试过该代码，因为我没有模型，因此它可能无法直接工作。

import multiprocessing as mp


# if everything is independent (eg. "infectionList" is static and does not change during the iterations)

def worker(model, infectionList):
    sirs = []
    for it in model.get_nodes():
        sir = model.infected_SIR_MODEL(it, infectionList, False)
        sirs.append(sir)
    return sirs

count = 500
infectionList = []
model = "YOUR MODEL INSTANCE"

data = [(model, infectionList) for _ in range(1, count+1)]
with mp.Pool() as pool:
    results = pool.starmap(worker, data)

如果“infectionList”或其他内容在“i”的每次迭代中更新，则提出第二个解决方案:

def worker2(model, it, infectionList):
    sir = model.infected_SIR_MODEL(it, infectionList, False)
    return sir

with mp.Pool() as pool:
    for i in range(1, count+1):
        data = [(model, it, infectionList) for it in model.get_nodes()]
        results = pool.starmap(worker2, data)

        # process results, update something go to next iteration....

编辑:更清楚地更新了单独提案的答案。