本文目录导读：
我来介绍使用NetworkX进行网络压缩的几种常用方法：
基于度的节点筛选
import networkx as nx
import matplotlib.pyplot as plt
import random
# 创建示例图
G = nx.barabasi_albert_graph(100, 2) # 无标度网络
# 方法1：按节点度筛选
def compress_by_degree(G, threshold=5):
"""保留度大于阈值的节点"""
nodes_to_keep = [n for n, d in G.degree() if d > threshold]
G_compressed = G.subgraph(nodes_to_keep).copy()
return G_compressed
G_degree = compress_by_degree(G, threshold=10)
print(f"原始节点数: {G.number_of_nodes()}, 压缩后: {G_degree.number_of_nodes()}")
社区检测压缩
# 方法2：基于社区检测进行压缩
def compress_by_community(G, method='louvain'):
"""使用社区检测合并节点"""
try:
import community as community_louvain
# 使用Louvain算法
partition = community_louvain.best_partition(G)
except ImportError:
# 使用Girvan-Newman
from networkx.algorithms.community import girvan_newman
communities = girvan_newman(G)
first_level_communities = next(communities)
partition = {}
for i, comm in enumerate(first_level_communities):
for node in comm:
partition[node] = i
# 创建压缩图
G_compressed = nx.Graph()
# 为每个社区创建超级节点
community_nodes = {}
for node, community_id in partition.items():
if community_id not in community_nodes:
community_nodes[community_id] = []
community_nodes[community_id].append(node)
# 添加超级节点
for comm_id in community_nodes:
G_compressed.add_node(f"Community_{comm_id}",
size=len(community_nodes[comm_id]))
# 添加边（社区间的连接）
for (u, v) in G.edges():
comm_u = partition[u]
comm_v = partition[v]
if comm_u != comm_v:
G_compressed.add_edge(f"Community_{comm_u}",
f"Community_{comm_v}")
return G_compressed, partition
G_community, partition = compress_by_community(G)
print(f"社区数量: {len(G_community.nodes())}")
骨架提取（生成树）
# 方法3：使用最小生成树或最大生成树
def compress_by_spanning_tree(G, weight='weight', mode='maximum'):
"""使用生成树压缩网络"""
# 如果没有权重，添加随机权重
if not weight in G.edges(data=True)[0]:
for (u, v) in G.edges():
G[u][v][weight] = random.random()
if mode == 'maximum':
# 最大生成树（保留重要连接）
G_compressed = nx.maximum_spanning_tree(G, weight=weight)
else:
# 最小生成树（保留紧密连接）
G_compressed = nx.minimum_spanning_tree(G, weight=weight)
return G_compressed
G_tree = compress_by_spanning_tree(G, mode='maximum')
print(f"压缩后边数: {G_tree.number_of_edges()}")
基于介数中心性的压缩
# 方法4：保留重要节点
def compress_by_centrality(G, percentile=20):
"""保留中心性最高的节点"""
# 计算介数中心性
betweenness = nx.betweenness_centrality(G)
# 确定阈值
threshold = sorted(betweenness.values(), reverse=True)[
min(len(betweenness)-1, int(len(betweenness) * percentile / 100))
]
# 保留重要节点
nodes_to_keep = [n for n, c in betweenness.items() if c >= threshold]
G_compressed = G.subgraph(nodes_to_keep).copy()
return G_compressed
G_centrality = compress_by_centrality(G, percentile=30)
print(f"原始节点数: {G.number_of_nodes()}, 压缩后: {G_centrality.number_of_nodes()}")
综合压缩示例
# 完整的压缩流程示例
def complete_compression_pipeline(G, methods=['degree', 'community', 'tree'],
degree_threshold=10, percentile=30):
"""综合使用多种压缩方法"""
results = {}
for method in methods:
if method == 'degree':
# 基于度的压缩
compressed = compress_by_degree(G.copy(), degree_threshold)
results['degree'] = {
'graph': compressed,
'nodes': compressed.number_of_nodes(),
'edges': compressed.number_of_edges()
}
elif method == 'community':
# 基于社区的压缩
compressed, _ = compress_by_community(G.copy())
results['community'] = {
'graph': compressed,
'nodes': compressed.number_of_nodes(),
'edges': compressed.number_of_edges()
}
elif method == 'tree':
# 生成树压缩
compressed = compress_by_spanning_tree(G.copy())
results['tree'] = {
'graph': compressed,
'nodes': compressed.number_of_nodes(),
'edges': compressed.number_of_edges(),
'type': 'tree'
}
elif method == 'centrality':
# 中心性压缩
compressed = compress_by_centrality(G.copy(), percentile)
results['centrality'] = {
'graph': compressed,
'nodes': compressed.number_of_nodes(),
'edges': compressed.number_of_edges()
}
return results
# 执行压缩
compression_results = complete_compression_pipeline(G)
# 可视化比较
def visualize_compression_results(original_G, results):
"""可视化压缩结果"""
fig, axes = plt.subplots(2, 3, figsize=(15, 10))
# 原始图
pos = nx.spring_layout(original_G)
nx.draw(original_G, pos, ax=axes[0, 0], node_size=20, alpha=0.6)
axes[0, 0].set_title(f"原始网络\n{original_G.number_of_nodes()} nodes\n"
f"{original_G.number_of_edges()} edges")
# 各种压缩结果
for i, (method, data) in enumerate(results.items()):
row = (i + 1) // 3
col = (i + 1) % 3
compressed_G = data['graph']
nx.draw(compressed_G, nx.spring_layout(compressed_G),
ax=axes[row, col], node_size=20, alpha=0.6)
axes[row, col].set_title(f"{method}压缩\n{data['nodes']} nodes\n"
f"{data['edges']} edges")
plt.tight_layout()
plt.show()
# 可视化结果
visualize_compression_results(G, compression_results)
# 打印统计信息
print("\n=== 网络压缩统计 ===")
print(f"{'方法':<15} {'节点数':<10} {'边数':<10} {'压缩率(%)':<12}")
print("-" * 50)
print(f"{'原始网络':<15} {G.number_of_nodes():<10} {G.number_of_edges():<10} {'100':<12}")
for method, data in compression_results.items():
compression_rate = (data['nodes'] / G.number_of_nodes()) * 100
print(f"{method:<15} {data['nodes']:<10} {data['edges']:<10} "
f"{compression_rate:<12.2f}")
应用场景示例
# 实际应用：社交网络压缩
def social_network_compression_demo():
"""社交网络压缩示例"""
# 创建模拟社交网络
G = nx.watts_strogatz_graph(200, 6, 0.3)
print("=== 社交网络压缩 ===")
# 1. 基于活动度压缩（保留活跃用户）
activity = {n: random.randint(1, 100) for n in G.nodes()}
nx.set_node_attributes(G, activity, 'activity')
active_users = [n for n in G.nodes() if G.nodes[n]['activity'] > 50]
G_active = G.subgraph(active_users).copy()
print(f"活跃用户网络: {G_active.number_of_nodes()} 用户, "
f"{G_active.number_of_edges()} 关系")
# 2. 社区结构压缩
G_community, partition = compress_by_community(G)
print(f"社区压缩网络: {G_community.number_of_nodes()} 社区, "
f"{G_community.number_of_edges()} 社区间连接")
return G_active, G_community
# 运行示例
G_active, G_community = social_network_compression_demo()
这些方法可以根据具体需求选择使用,常见的应用场景包括：
- 数据可视化：减少网络复杂度
- 算法优化：降低计算复杂度
- 模式发现：提取网络核心结构
- 存储优化：节省存储空间
选择压缩方法时需要考虑：
- 网络类型和特征
- 保留信息的重要性
- 计算资源限制
- 应用具体需求
