基于ray 多进程调度管理能力优化networks节点最短路径的并行计算

deff_one():
degree_centrality_result=betweenness_centrality(G)
out= []
fori, jindegree_centrality_result.items():
out.append({"user_id": i, "weight": j})
pd.DataFrame(out).to_csv("betweenness_centrality_result.csv")
# 读取2020年企业年报因果关系数据集。bracket_name=json.load(open("2020_extract_result_center.json", "r"))
graph= {}
foriintqdm(list(bracket_name)):
if (i[0], i[4]) notingraph:
graph[(i[0], i[4])] =1else:
graph[(i[0], i[4])] +=1print(len(graph))
graph_relation= []
forgraph_id, graph_countingraph.items():
ifgraph_count>1:
graph_relation.append(graph_id)
print(len(graph_relation))
graph_relation=graph_relation[:10000]
G=nx.Graph()
G.add_edges_from(graph_relation)  # 添加多条边ray_G=ray.put(G)
f_one()

importjsonimportnetworkxasnximportpandasaspdfromcollectionsimportdequefromnetworkx.algorithms.shortest_paths.weightedimport_weight_functionfromtqdmimporttqdm# helpers for betweenness centralitydef_single_source_shortest_path_basic(G, s):
S= []
P= {}
forvinG:
P[v] = []
sigma=dict.fromkeys(G, 0.0)  # sigma[v]=0 for v in GD= {}
sigma[s] =1.0D[s] =0Q=deque([s])
whileQ:  # use BFS to find shortest pathsv=Q.popleft()
S.append(v)
Dv=D[v]
sigmav=sigma[v]
forwinG[v]:
ifwnotinD:
Q.append(w)
D[w] =Dv+1ifD[w] ==Dv+1:  # this is a shortest path, count pathssigma[w] +=sigmavP[w].append(v)  # predecessorsreturnS, P, sigma, Ddef_accumulate_basic(betweenness, S, P, sigma, s):
delta=dict.fromkeys(S, 0)
whileS:
w=S.pop()
coeff= (1+delta[w]) /sigma[w]
forvinP[w]:
delta[v] +=sigma[v] *coeffifw!=s:
betweenness[w] +=delta[w]
returnbetweenness, deltadef_accumulate_endpoints(betweenness, S, P, sigma, s):
betweenness[s] +=len(S) -1delta=dict.fromkeys(S, 0)
whileS:
w=S.pop()
coeff= (1+delta[w]) /sigma[w]
forvinP[w]:
delta[v] +=sigma[v] *coeffifw!=s:
betweenness[w] +=delta[w] +1returnbetweenness, deltadef_accumulate_edges(betweenness, S, P, sigma, s):
delta=dict.fromkeys(S, 0)
whileS:
w=S.pop()
coeff= (1+delta[w]) /sigma[w]
forvinP[w]:
c=sigma[v] *coeffif (v, w) notinbetweenness:
betweenness[(w, v)] +=celse:
betweenness[(v, w)] +=cdelta[v] +=cifw!=s:
betweenness[w] +=delta[w]
returnbetweennessdef_rescale(betweenness, n, normalized, directed=False, k=None, endpoints=False):
ifnormalized:
ifendpoints:
ifn<2:
scale=None# no normalizationelse:
# Scale factor should include endpoint nodesscale=1/ (n* (n-1))
elifn<=2:
scale=None# no normalization b=0 for all nodeselse:
scale=1/ ((n-1) * (n-2))
else:  # rescale by 2 for undirected graphsifnotdirected:
scale=0.5else:
scale=NoneifscaleisnotNone:
ifkisnotNone:
scale=scale*n/kforvinbetweenness:
betweenness[v] *=scalereturnbetweennessdef_rescale_e(betweenness, n, normalized, directed=False, k=None):
ifnormalized:
ifn<=1:
scale=None# no normalization b=0 for all nodeselse:
scale=1/ (n* (n-1))
else:  # rescale by 2 for undirected graphsifnotdirected:
scale=0.5else:
scale=NoneifscaleisnotNone:
ifkisnotNone:
scale=scale*n/kforvinbetweenness:
betweenness[v] *=scalereturnbetweennessimportrayray.init()
@ray.remotedefshortest_path_basic(s, ray_betweenness=None):
G=ray.get(ray_G)
betweenness=dict.fromkeys(G, 0.0)  # b[v]=0 for v in G# single source shortest pathsS, P, sigma, _=_single_source_shortest_path_basic(G, s)
# accumulationbetweenness, delta=_accumulate_basic(betweenness, S, P, sigma, s)
#     ray_betweenness = ray.put(betweenness)delGreturnbetweennessdefbetweenness_centrality(
G, k=None, normalized=True, weight=None, endpoints=False, seed=None):
betweenness=dict.fromkeys(G, 0.0)  # b[v]=0 for v in GifkisNone:
nodes=Gelse:
nodes=seed.sample(list(G.nodes()), k)
ray_betweenness=ray.put(betweenness)
ray_betweenness= [shortest_path_basic.remote(s, ray_betweenness) forsinnodes]
betweenness=ray.get(ray_betweenness)
#     print(betweenness[:3])# rescaling#     json.dump(betweenness, open("betweenness.json", "w", encoding="utf-8"), ensure_ascii=False)betweenness_compute= {}
forbetweenness_oneinbetweenness:
fori,jinbetweenness_one.items():
ifiinbetweenness_compute:
betweenness_compute[i]+=jelse:
betweenness_compute[i]=jbetweenness=_rescale(
betweenness_compute,
len(G),
normalized=normalized,
directed=G.is_directed(),
k=k,
endpoints=endpoints,
    )
returnbetweenness@ray.remotedeff(index):
bracket_name=ray.get(bracket_name_ray)
graph= {}
print(index)
foriintqdm(list(bracket_name)):
if (i[0], i[2]) notingraph:
graph[(i[0], i[2])] =1else:
graph[(i[0], i[2])] +=1print(len(graph))
graph_relation= []
forgraph_id, graph_countingraph.items():
ifgraph_count>1:
graph_relation.append(graph_id)
G=nx.Graph()
G.add_edges_from(graph_relation)  # 添加多条边degree_centrality_result=betweenness_centrality(G)
out= []
fori, jindegree_centrality_result.items():
out.append({"user_id": i, "weight": j})
pd.DataFrame(out).to_csv("betweenness_centrality_result"+str(index) +".csv")
deff_one():
degree_centrality_result=betweenness_centrality(G)
out= []
fori, jindegree_centrality_result.items():
out.append({"user_id": i, "weight": j})
pd.DataFrame(out).to_csv("betweenness_centrality_result.csv")
bracket_name=json.load(open("2020_extract_result_center.json", "r"))
bracket_name_ray=ray.put(bracket_name)
# futures = [f.remote(i) for i in range(len(bracket_name) // 3000)]# print(ray.get(futures))  # [0, 1, 4, 9]bracket_name=ray.get(bracket_name_ray)
graph= {}
foriintqdm(list(bracket_name)):
if (i[0], i[4]) notingraph:
graph[(i[0], i[4])] =1else:
graph[(i[0], i[4])] +=1print(len(graph))
graph_relation= []
forgraph_id, graph_countingraph.items():
ifgraph_count>1:
graph_relation.append(graph_id)
print(len(graph_relation))
graph_relation=graph_relation[:10000]
G=nx.Graph()
G.add_edges_from(graph_relation)  # 添加多条边ray_G=ray.put(G)
f_one()