Dijkstra算法复杂度评估：如何从CSV数据集随机选取不同规模子集

Hey there! Let's walk through exactly how to generate those test subsets, time your Dijkstra implementation, and plot the Big-O complexity graph. You're already thinking in the right direction by testing with increasing input sizes—great call!

1. 读取CSV并构建图结构 #

First, we need to turn your city-distance CSV into a graph structure your Dijkstra algorithm can work with (an adjacency list is perfect here). Pandas makes reading the CSV straightforward:

import pandas as pd
import random
import time
import matplotlib.pyplot as plt
import heapq

# 读取你的CSV数据
df = pd.read_csv('your_city_dataset.csv')

# 构建无向图的邻接表
graph = {}
for _, row in df.iterrows():
    city1, city2, distance = row['city1'], row['city2'], row['distance']
    # 双向添加边（假设城市间的道路是双向的）
    if city1 not in graph:
        graph[city1] = []
    graph[city1].append((city2, distance))
    if city2 not in graph:
        graph[city2] = []
    graph[city2].append((city1, distance))

2. 生成不同大小的随机测试子集 #

You have two main options here—pick the one that fits your testing needs:

方法1：简单随机子集（快速但可能有孤立节点） #

This directly samples rows from the CSV, but could create disconnected subgraphs (which might break pathfinding for some nodes):

def get_random_subset(df, row_count):
    # 随机选取指定行数的子集
    subset_df = df.sample(n=row_count, random_state=42)  # random_state保证结果可复现
    # 从子集重建邻接表
    subset_graph = {}
    for _, row in subset_df.iterrows():
        c1, c2, d = row['city1'], row['city2'], row['distance']
        if c1 not in subset_graph:
            subset_graph[c1] = []
        subset_graph[c1].append((c2, d))
        if c2 not in subset_graph:
            subset_graph[c2] = []
        subset_graph[c2].append((c1, d))
    return subset_graph, len(subset_graph)  # 返回子集图和节点数量

方法2：连通子集（更推荐，保证测试有效性） #

This ensures your subset is a connected graph, so Dijkstra can calculate paths between any pair of nodes:

def get_connected_subset(full_graph, target_node_count):
    # 随机选一个起始城市
    start_city = random.choice(list(full_graph.keys()))
    connected_cities = {start_city}
    
    # 逐步添加相邻城市，直到达到目标节点数
    while len(connected_cities) < target_node_count:
        current_city = random.choice(list(connected_cities))
        neighbors = [neighbor for neighbor, _ in full_graph[current_city]]
        # 筛选未加入连通集合的邻居
        available_neighbors = [n for n in neighbors if n not in connected_cities]
        if available_neighbors:
            new_city = random.choice(available_neighbors)
            connected_cities.add(new_city)
        else:
            # 若当前城市无新邻居，换一个重试
            continue
    
    # 构建连通子集的邻接表
    subset_graph = {city: [] for city in connected_cities}
    for city in connected_cities:
        for neighbor, distance in full_graph[city]:
            if neighbor in connected_cities:
                subset_graph[city].append((neighbor, distance))
    return subset_graph, len(connected_cities)

3. 计时运行你的Dijkstra算法 #

Assuming you have your custom Dijkstra implementation, we'll time it multiple times per subset to reduce measurement error:

首先，你的Dijkstra示例（替换成你自己的实现） #

def your_dijkstra(graph, start):
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    priority_queue = [(0, start)]
    visited = set()
    
    while priority_queue:
        current_dist, current_node = heapq.heappop(priority_queue)
        if current_node in visited:
            continue
        visited.add(current_node)
        for neighbor, weight in graph[current_node]:
            if distances[neighbor] > current_dist + weight:
                distances[neighbor] = current_dist + weight
                heapq.heappush(priority_queue, (distances[neighbor], neighbor))
    return distances

计时函数 #

def measure_average_time(graph, runs=5):
    # 随机选一个起始节点（确保在子集图中）
    start_node = random.choice(list(graph.keys()))
    total_time = 0.0
    # 多次运行取平均，抵消系统负载影响
    for _ in range(runs):
        start_time = time.perf_counter()
        your_dijkstra(graph, start_node)
        end_time = time.perf_counter()
        total_time += (end_time - start_time)
    return total_time / runs

4. 收集数据并绘制Big-O图 #

Now we'll run tests on 5,10,15,20-node subsets and plot the results against theoretical Big-O curves:

# 测试的目标节点规模
target_sizes = [5, 10, 15, 20]
average_times = []
actual_node_counts = []

for size in target_sizes:
    # 替换成get_random_subset如果你想用简单随机方法
    subset_graph, node_count = get_connected_subset(graph, size)
    avg_time = measure_average_time(subset_graph)
    average_times.append(avg_time)
    actual_node_counts.append(node_count)
    print(f"节点数: {node_count}, 平均耗时: {avg_time:.6f}秒")

# 绘制耗时 vs 节点数的折线图
plt.figure(figsize=(10, 6))
plt.plot(actual_node_counts, average_times, 'bo-', label='你的Dijkstra实现')

# 可选：添加理论复杂度曲线（比如O(n log n)，适配你的数据缩放）
import numpy as np
n = np.array(actual_node_counts)
# 计算缩放系数，让理论曲线和实际数据对齐
scale_factor = average_times[-1] / (n[-1] * np.log2(n[-1]))
plt.plot(n, scale_factor * n * np.log2(n), 'r--', label='理论O(n log n)')

plt.xlabel('节点数量')
plt.ylabel('平均耗时（秒）')
plt.title('Dijkstra算法复杂度测试')
plt.legend()
plt.grid(True)
plt.show()

小贴士 #

• 可复现性：固定random_state或设置random.seed(42)，这样每次测试的子集完全一致，便于对比迭代改进
• 复杂度适配：如果你的Dijkstra用的是普通队列而非优先队列，理论复杂度是O(n²)，把理论曲线换成scale_factor * n**2即可
• 边数替代：Dijkstra的实际复杂度和边数E也相关，你可以记录每个子集的边数，用E作为X轴绘图

内容的提问来源于stack exchange，提问作者foxy15