python - 使用并行 python 并行化 networkx 中节点的循环

Question

我正在 networkx 中做一些复杂的计算。计算涉及为网络中的每个节点一次又一次地计算一些数量。作为此类计算的一个示例，假设我们要计算网络中每个节点的平均邻居度并将该值保存为节点属性。以下代码段对我有用:

import networkx as nx

G = nx.erdos_renyi_graph(10, 0.5)

def ave_nbr_deg(node):
    value = 0.
    for nbr in G.neighbors(node):
        value += G.degree(nbr)
    G.node[node]['ave_nbr_deg'] = value/len(G.neighbors(node))

for node in G.nodes():
    ave_nbr_deg(node)

print G.nodes(data = True)

这给了我:

[(0, {'ave_nbr_deg': 5.4}), (1, {'ave_nbr_deg': 5.0}), (2, {'ave_nbr_deg': 5.333333333333333}), (3, {'ave_nbr_deg': 5.2}), (4, {'ave_nbr_deg': 5.6}), (5, {'ave_nbr_deg': 5.6}), (6, {'ave_nbr_deg': 5.2}), (7, {'ave_nbr_deg': 5.25}), (8, {'ave_nbr_deg': 5.5}), (9, {'ave_nbr_deg': 5.5})]

这里我自己有一个小疑问。对象 G 是在函数 ave_nbr_deg 之外创建的，我不知道该函数是如何获得其信息的，即使我没有将其声明为全局函数。

现在，我想使用并行 python 模块来使用我系统上的所有内核来进行此计算。对上述代码进行一些更改后，我得到以下代码:

import networkx as nx
import pp

G = nx.erdos_renyi_graph(10, 0.5)

def ave_nbr_deg(node):
    value = 0.
    for nbr in G.neighbors(node):
        value += G.degree(nbr)
    G.node[node]['ave_nbr_deg'] = value/len(G.neighbors(node))

job_server = pp.Server(ppservers = ())

print "Starting pp with", job_server.get_ncpus(), "workers"

for node in G.nodes():
    job_server.submit(ave_nbr_deg, args = (node,))()

print G.nodes(data = True)

但它返回以下错误:

An error has occured during the function execution
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/ppworker.py", line 90, in run
    __result = __f(*__args)
  File "<string>", line 5, in ave_nbr_deg
NameError: global name 'G' is not defined

我尝试了各种方法，包括模块名称 nx 提交等等。但是，我没有得到确切的问题。不幸的是，documentation of smp太短无法解决这个问题。如果这里有人可以帮助我，我将不胜感激。

提前致谢

Answer 1

回答您的第一个问题:Python 首先在本地命名空间中查找变量。如果它在那里没有找到它们，那么它将向上移动到父命名空间并在那里寻找它。这就是它找到 G 的方式，即使 G 没有在本地声明。以下是有关变量范围的更多信息:Python scoping

大概有两种解决方法:

1 - 将 G 作为参数传递给函数 ave_nbr_deg，例如

for node in G.nodes():
    job_server.submit(ave_nbr_deg, args = (node,G))()

2 - 在提交参数中将 G 声明为全局变量(来自 Parallel Python 文档)

   submit(self, func, args=(), depfuncs=(), modules=(), 
       callback=None, callbackargs=(), group='default', globals=None)
Submits function to the execution queue

func - function to be executed
args - tuple with arguments of the 'func'
depfuncs - tuple with functions which might be called from 'func'
modules - tuple with module names to import
callback - callback function which will be called with argument 
        list equal to callbackargs+(result,) 
        as soon as calculation is done
callbackargs - additional arguments for callback function
group - job group, is used when wait(group) is called to wait for
jobs in a given group to finish
globals - dictionary from which all modules, functions and classes
will be imported, for instance: globals=globals()

在这种情况下，调用以下内容应该有效:

for node in G.nodes():
    job_server.submit(ave_nbr_deg, args = (node,G), globals=globals())()